note

• 通过编排动作设置哲学家智能体的"示例任务"，目的是让 Agent 更好地理解如何回答问题。主要包括设置示例问题、定义思考过程、应用到所有哲学家。建立了一个"先思考，后总结"的回答模式，这种方式相当于给AI提供了一个"样板"，告诉它：“这就是我们期望你回答问题的方式”

一、构建智能搜索代理

• Zigent 是一个基于 Agentlite 框架改进的智能代理开发框架。Agentlite 最初由 Salesforce AI Research 团队开发。Zigent的参考地址：https://github.com/datawhalechina/wow-agent
• 我们要搞的这个搜索代理能够通过 DuckDuckGo或Bocha 搜索引擎查找信息并回答问题。

DuckSearchAction类的两个作用：

• 初始化时配置动作的名称、描述和参数说明
• 通过 call 方法执行实际的搜索操作 使用示例见代码中的search_action部分

代码部分：

import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()
# 初始化变量
base_url = None
chat_model = None
api_key = None

# 使用with语句打开文件，确保文件使用完毕后自动关闭
env_path = "/Users/guomiansheng/Desktop/LLM/llm_app/wow-agent/.env.txt"
with open(env_path, 'r') as file:
    # 逐行读取文件
    for line in file:
        # 移除字符串头尾的空白字符（包括'\n'）
        line = line.strip()
        # 检查并解析变量
        if "base_url" in line:
            base_url = line.split('=', 1)[1].strip().strip('"')
        elif "chat_model" in line:
            chat_model = line.split('=', 1)[1].strip().strip('"')
        elif "ZHIPU_API_KEY" in line:
            api_key = line.split('=', 1)[1].strip().strip('"')
        elif "BOCHA_API_KEY" in line:
            BOCHA_API_KEY = line.split('=', 1)[1].strip().strip('"')

# 打印变量以验证
print(f"base_url: {base_url}")
print(f"chat_model: {chat_model}")
print(f"ZHIPU_API_KEY: {api_key}")


from typing import List
from zigent.agents import ABCAgent, BaseAgent
from zigent.llm.agent_llms import LLM
from zigent.commons import TaskPackage
from zigent.actions.BaseAction import BaseAction
from zigent.logging.multi_agent_log import AgentLogger
from duckduckgo_search import DDGS

llm = LLM(api_key=api_key, base_url=base_url, model_name=chat_model)
response = llm.run("你是谁？")
print(response)


# 一、创建搜索动作
class DuckSearchAction(BaseAction):
    def __init__(self) -> None:
        action_name = "DuckDuckGo_Search"
        action_desc = "Using this action to search online content."
        params_doc = {"query": "the search string. be simple."}
        self.ddgs = DDGS()
        super().__init__(
            action_name=action_name, 
            action_desc=action_desc, 
            params_doc=params_doc,
        )

    def __call__(self, query):
        results = self.ddgs.chat(query)
        return results
    

# search_action = DuckSearchAction()
# results = search_action("什么是 agent")
# print(results)


# BOCHA_API_KEY = os.getenv('BOCHA_API_KEY')
import requests
import json
# 定义Bocha Web Search工具
def bocha_web_search_tool(query: str, count: int = 8) -> str:
    """
    使用Bocha Web Search API进行联网搜索，返回搜索结果的字符串。
    
    参数:
    - query: 搜索关键词
    - count: 返回的搜索结果数量

    返回:
    - 搜索结果的字符串形式
    """
    url = 'https://api.bochaai.com/v1/web-search'
    headers = {
        'Authorization': f'Bearer {BOCHA_API_KEY}',  # 请替换为你的API密钥
        'Content-Type': 'application/json'
    }
    data = {
        "query": query,
        "freshness": "noLimit", # 搜索的时间范围，例如 "oneDay", "oneWeek", "oneMonth", "oneYear", "noLimit"
        "summary": True, # 是否返回长文本摘要总结
        "count": count
    }

    response = requests.post(url, headers=headers, json=data)

    if response.status_code == 200:
        # 返回给大模型的格式化的搜索结果文本
        # 可以自己对博查的搜索结果进行自定义处理
        return json.dumps(response.json())
    else:
        raise Exception(f"API请求失败，状态码: {response.status_code}, 错误信息: {response.text}")



class BochaSearchAction(BaseAction):
    def __init__(self) -> None:
        action_name = "Bocha_Search"
        action_desc = "Using this action to search online content."
        params_doc = {"query": "the search string. be simple."}
        super().__init__(
            action_name=action_name, 
            action_desc=action_desc, 
            params_doc=params_doc,
        )

    def __call__(self, query):
        results = bocha_web_search_tool(query)
        rst = json.loads(results)
        result = ""
        for item in rst["data"]["webPages"]["value"]:
            result += item["summary"]
        return result


search_action = BochaSearchAction()
results = search_action("上海有哪些私立小学在招聘小学英语老师？")
print(results)


# 二、创建搜索代理
class DuckSearchAgent(BaseAgent):
    def __init__(
        self,
        llm: LLM,
        actions: List[BaseAction] = [DuckSearchAction()],
        manager: ABCAgent = None,
        **kwargs
    ):
        name = "duck_search_agent"
        role = "You can answer questions by using duck duck go search content."
        super().__init__(
            name=name,
            role=role,
            llm=llm,
            actions=actions,
            manager=manager,
            logger=agent_logger,
        )


# search_action = BochaSearchAction()
class BochaSearchAgent(BaseAgent):
    def __init__(
        self,
        llm: LLM,
        actions: List[BaseAction] = [search_action],
        manager: ABCAgent = None,
        **kwargs
    ):
        name = "bocha_search_agent"
        role = "You can answer questions by using bocha search content."
        super().__init__(
            name=name,
            role=role,
            llm=llm,
            actions=actions,
            manager=manager,
        )


# 三、执行代理
# def do_search_agent():
#     # 创建代理实例
#     search_agent = DuckSearchAgent(llm=llm)

#     # 创建任务
#     task = "what is the found date of microsoft"
#     task_pack = TaskPackage(instruction=task)

#     # 执行任务并获取响应
#     response = search_agent(task_pack)
#     print("response:", response)

# if __name__ == "__main__":
#     do_search_agent()

def do_search_agent():
    # 创建代理实例
    search_agent = BochaSearchAgent(llm=llm)

    # 创建任务
    task = "what is the found date of microsoft"
    task_pack = TaskPackage(instruction=task)

    # 执行任务并获取响应
    response = search_agent(task_pack)
    print("response:", response)


if __name__ == "__main__":
    do_search_agent()

通过两步的action和observation，最后也回答出了微软的成立时间：

二、构建哲学家多智能体

• 实现一个 Philosopher 类，继承自 BaseAgent 基类，通过构造函数接收哲学家名字、语言模型等参数，设置角色提示词,让AI扮演特定哲学家的角色。每个哲学家智能体都有自己的名字和角色定位，使用相同的语言模型(llm)，可以根据自己的哲学思想发表观点。这样的设计让我们可以模拟不同哲学家之间的对话和思想交流
• 通过编排动作设置哲学家智能体的"示例任务"，目的是让 Agent 更好地理解如何回答问题。主要包括设置示例问题、定义思考过程、应用到所有哲学家。建立了一个"先思考，后总结"的回答模式，这种方式相当于给AI提供了一个"样板"，告诉它：“这就是我们期望你回答问题的方式”
• 实现一个多智能体系统中的管理者代理（Manager Agent），负责协调孔子、苏格拉底和亚里士多德三位哲学家。管理者的角色定义为：依次询问各位哲学家的观点并总结。ManagerAgent 中设置了"什么是生命的意义？"示例任务流程，包括思考如何处理任务、询问哲学家的观点、总结观点等步骤
• 最后调用 ManagerAgent。管理者 Agent 会根据我们之前设置的示例任务和动作序列,按照类似的思路来处理新的任务。在这个例子中，我们设置了一个新的任务"先有鸡还是先有蛋?"，然后调用管理者 Agent 来处理这个任务

import os
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()
# 初始化变量
base_url = None
chat_model = None
api_key = None

# 使用with语句打开文件，确保文件使用完毕后自动关闭
env_path = "/Users/guomiansheng/Desktop/LLM/llm_app/wow-agent/.env.txt"
with open(env_path, 'r') as file:
    # 逐行读取文件
    for line in file:
        # 移除字符串头尾的空白字符（包括'\n'）
        line = line.strip()
        # 检查并解析变量
        if "base_url" in line:
            base_url = line.split('=', 1)[1].strip().strip('"')
        elif "chat_model" in line:
            chat_model = line.split('=', 1)[1].strip().strip('"')
        elif "ZHIPU_API_KEY" in line:
            api_key = line.split('=', 1)[1].strip().strip('"')
        elif "BOCHA_API_KEY" in line:
            BOCHA_API_KEY = line.split('=', 1)[1].strip().strip('"')

# 打印变量以验证
print(f"base_url: {base_url}")
print(f"chat_model: {chat_model}")
print(f"ZHIPU_API_KEY: {api_key}")


from openai import OpenAI
client = OpenAI(
    api_key = api_key,
    base_url = base_url
)
print(client)

def get_completion(prompt):
    response = client.chat.completions.create(
        model="glm-4-flash",  # 填写需要调用的模型名称
        messages=[
            {"role": "user", "content": prompt},
        ],
    )
    return response.choices[0].message.content


# 一、使用zigent定义llm
from typing import List
from zigent.agents import ABCAgent, BaseAgent
from zigent.llm.agent_llms import LLM
from zigent.commons import TaskPackage
from zigent.actions.BaseAction import BaseAction
from zigent.logging.multi_agent_log import AgentLogger
from duckduckgo_search import DDGS

llm = LLM(api_key=api_key, base_url=base_url, model_name=chat_model)
response = llm.run("你是谁？")
print(response)


# 二、哲学家
from typing import List
from zigent.actions.BaseAction import BaseAction
from zigent.agents import ABCAgent, BaseAgent

# 定义 Philosopher 类，继承自 BaseAgent 类
class Philosopher(BaseAgent):
    def __init__(
        self,
        philosopher,
        llm: LLM, # BaseLLM,
        actions: List[BaseAction] = [], 
        manager: ABCAgent = None,
        **kwargs
    ):
        name = philosopher
        # 角色
        role = f"""You are {philosopher}, the famous educator in history. You are very familiar with {philosopher}'s Book and Thought. Tell your opinion on behalf of {philosopher}."""
        super().__init__(
            name=name,
            role=role,
            llm=llm,
            actions=actions,
            manager=manager
        )

# 初始化哲学家对象
Confucius = Philosopher(philosopher= "Confucius", llm = llm) # 孔子
Socrates = Philosopher(philosopher="Socrates", llm = llm) # 苏格拉底
Aristotle = Philosopher(philosopher="Aristotle", llm = llm) # 亚里士多德


# 三、编排动作
# 导入必要的模块
from zigent.commons import AgentAct, TaskPackage
from zigent.actions import ThinkAct, FinishAct
from zigent.actions.InnerActions import INNER_ACT_KEY
from zigent.agents.agent_utils import AGENT_CALL_ARG_KEY

# 为哲学家智能体添加示例任务
# 设置示例任务:询问生命的意义
exp_task = "What do you think the meaning of life?"
exp_task_pack = TaskPackage(instruction=exp_task)

# 第一个动作:思考生命的意义
act_1 = AgentAct(
    name=ThinkAct.action_name,
    params={INNER_ACT_KEY: f"""Based on my thought, we are born to live a meaningful life, and it is in living a meaningful life that our existence gains value. Even if a life is brief, if it holds value, it is meaningful. A life without value is merely existence, a mere survival, a walking corpse."""
    },
)
# 第一个动作的观察结果
obs_1 = "OK. I have finished my thought, I can pass it to the manager now."

# 第二个动作:总结思考结果
act_2 = AgentAct(name=FinishAct.action_name, params={INNER_ACT_KEY: "I can summarize my thought now."})
# 第二个动作的观察结果
obs_2 = "I finished my task, I think the meaning of life is to pursue value for the whold world."
# 将动作和观察组合成序列
exp_act_obs = [(act_1, obs_1), (act_2, obs_2)]

# 为每个哲学家智能体添加示例
# 为孔子添加示例
Confucius.prompt_gen.add_example(
    task = exp_task_pack, action_chain = exp_act_obs
)
# 为苏格拉底添加示例
Socrates.prompt_gen.add_example(
    task = exp_task_pack, action_chain = exp_act_obs
)
# 为亚里士多德添加示例
Aristotle.prompt_gen.add_example(
    task = exp_task_pack, action_chain = exp_act_obs
)


# 四、定义 ManagerAgent
# 定义管理者代理
from zigent.agents import ManagerAgent

# 设置管理者代理的基本信息
manager_agent_info = {
    "name": "manager_agent",
    "role": "you are managing Confucius, Socrates and Aristotle to discuss on questions. Ask their opinion one by one and summarize their view of point."
}
# 设置团队成员
team = [Confucius, Socrates, Aristotle]
# 创建管理者代理实例
manager_agent = ManagerAgent(name=manager_agent_info["name"], role=manager_agent_info["role"], llm=llm, TeamAgents=team)

# 为管理者代理添加示例任务
exp_task = "What is the meaning of life?"
exp_task_pack = TaskPackage(instruction=exp_task)

# 第一步：思考如何处理任务
act_1 = AgentAct(
    name=ThinkAct.action_name,
    params={INNER_ACT_KEY: f"""I can ask Confucius, Socrates and Aristotle one by one on their thoughts, and then summary the opinion myself."""
    },
)
obs_1 = "OK."

# 第二步：询问孔子的观点
act_2 = AgentAct(
    name=Confucius.name,
    params={AGENT_CALL_ARG_KEY: "What is your opinion on the meaning of life?",
        },
)
obs_2 = """Based on my thought, I think the meaning of life is to pursue value for the whold world."""

# 第三步：思考下一步行动
act_3 = AgentAct(
    name=ThinkAct.action_name,
    params={INNER_ACT_KEY: f"""I have obtained information from Confucius, I need to collect more information from Socrates."""
    },
)
obs_3 = "OK."

# 第四步：询问苏格拉底的观点
act_4 = AgentAct(
    name=Socrates.name,
    params={AGENT_CALL_ARG_KEY: "What is your opinion on the meaning of life?",
        },
)
obs_4 = """I think the meaning of life is finding happiness."""

# 第五步：继续思考下一步
act_5 = AgentAct(
    name=ThinkAct.action_name,
    params={INNER_ACT_KEY: f"""I have obtained information from Confucius and Socrates, I can collect more information from Aristotle."""
    },
)
obs_5 = "OK."

# 第六步：询问亚里士多德的观点
act_6 = AgentAct(
    name=Aristotle.name,
    params={AGENT_CALL_ARG_KEY: "What is your opinion on the meaning of life?",
        },
)
obs_6 = """I believe the freedom of spirit is the meaning."""

# 最后一步：总结所有观点
act_7 = AgentAct(name=FinishAct.action_name, params={INNER_ACT_KEY: "Their thought on the meaning of life is to pursue value, happiniss and freedom of spirit."})
obs_7 = "Task Completed. The meaning of life is to pursue value, happiness and freedom of spirit."

# 将所有动作和观察组合成序列
exp_act_obs = [(act_1, obs_1), (act_2, obs_2), (act_3, obs_3), (act_4, obs_4), (act_5, obs_5), (act_6, obs_6), (act_7, obs_7)]

# 将示例添加到管理者代理的提示生成器中
manager_agent.prompt_gen.add_example(
    task = exp_task_pack, action_chain = exp_act_obs
)


# 五、调用 ManagerAgent
from zigent.commons import AgentAct, TaskPackage

exp_task = "先有鸡还是先有蛋?"
exp_task_pack = TaskPackage(instruction=exp_task)
manager_agent(exp_task_pack)

辩论的结果如下：

Reference

[1] https://github.com/datawhalechina/wow-agent
[2] https://www.datawhale.cn/learn/summary/86
[3] https://open.bochaai.com/
[4] 官方文档：https://docs.cloud.llamaindex.ai/
[5] 学习笔记：https://www.cnblogs.com/HYLOVEYOURSELF/p/18680532