实现多agent协作

多Agent协作不同的Agent有不同的能力我们可能会有各种实际需求例如实时识别车牌位置Yolo-识别车牌内容qwen-vl- LLM管理记录车牌信息。通过多Agent协作的工作流能够实现拍照答题自动剪辑ppt生成等一系列复杂问题。下面用一个简单的案例来说明。简单的多Agent协作示例需求查一下阿里、腾讯、百度的PE并计算平均值。import os import operator from pydantic import BaseModel, Field from langchain_community.chat_models.tongyi import ChatTongyi from langchain_core.tools import tool from langchain_core.messages import HumanMessage, BaseMessage, ToolMessage, SystemMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from typing import Annotated, List, Literal, TypedDict from langgraph.graph import StateGraph, END os.environ[DASHSCOPE_API_KEY] apikey llm ChatTongyi(modelqwen-plus) tool def web_search(query: str): 用于查找最新的股票数据、公司财报信息。 results [] if 阿里 in query: results.append(阿里巴巴(BABA) PE: 15.5) if 腾讯 in query: results.append(腾讯控股(0700) PE: 18.2) if 百度 in query: results.append(百度(BIDU) PE: 11.8) if not results: return 未找到数据 return ; .join(results) tool def python_calculator(code: str): 用于计算。输入必须是 python 代码。 try: result eval(code) return f计算结果: {result} except Exception as e: return f计算错误: {e} def create_agent(state: dict, llm, tools, system_prompt): llm_tools llm.bind_tools(tools) prompt [SystemMessage(contentsystem_prompt)] state[messages] response llm_tools.invoke(prompt) results [response] for tool_call in response.tool_calls: func_name tool_call[name] args tool_call[args] call_id tool_call[id] func next((t for t in tools if t.name func_name), None) if func: tool_output func.invoke(args) tool_msg ToolMessage( contentstr(tool_output), namefunc_name, tool_call_idcall_id ) results.append(tool_msg) return {messages: results} class State(TypedDict): messages: Annotated[List[BaseMessage], operator.add] next: str def researcher_node(state): return create_agent( statestate, llmllm, tools[web_search], system_prompt你是一个研究员。只负责查数据找到数据后直接输出原话不需要计算。 ) def coder_node(state): return create_agent( statestate, llmllm, tools[python_calculator], system_prompt你是一个程序员。根据上面研究员查到的数据写代码计算平均值。 ) def finish_node(state): return create_agent( statestate, llmllm, tools[], system_prompt任务完成简短的总结最终结果。 ) def supervisor_node(state): system_prompt ( 你是项目经理。根据对话历史决定下一步交给谁。 查数据找 Researcher计算找 Coder识别图片找 Photographer。 如果任务完成必须选择 FINISH。 ) prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namemessages), (system, 根据以上情况请做出选择。), ]) class RouteResponse(BaseModel): next: Literal[Researcher, Coder, FINISH] Field( ..., description下一步交给谁如果任务完成请选 FINISH ) chain prompt | llm.with_structured_output(RouteResponse) response chain.invoke(state) return {next: response.next} def init_agent(): workflow StateGraph(State) workflow.add_node(Researcher, researcher_node) workflow.add_node(Coder, coder_node) workflow.add_node(Supervisor, supervisor_node) workflow.add_node(Finish, finish_node) workflow.add_edge(Researcher, Supervisor) workflow.add_edge(Coder, Supervisor) workflow.add_edge(Finish, END) workflow.add_conditional_edges( Supervisor, lambda state: state[next],{ Researcher: Researcher, Coder: Coder, FINISH: Finish, }) workflow.set_entry_point(Supervisor) return workflow.compile() if __name__ __main__: agent init_agent() for result in agent.stream({ messages: [HumanMessage(content查一下阿里、腾讯、百度的PE并计算平均值。)] }): for key, value in result.items(): if key Supervisor: print([ key ] 去向: value[next]) else: print([ key ] 回复: value[messages][-1].content)代码解释代码一共用到了4个agentagent Researcher其有一个工具负责搜索某些内容agent coder其有一个工具负责进行精确计算agent finish其没有工具负责总结内容agent Supervisor其没有工具负责管理上面3个agent决定任务的去向上面案例使用的是langgraph组件这里就不详细讲解了请看之前文章。代码流程初始化工具库-初始化agent-构建图-运行初始化工具库tool def web_search(query: str): 用于查找最新的股票数据、公司财报信息。 results [] if 阿里 in query: results.append(阿里巴巴(BABA) PE: 15.5) if 腾讯 in query: results.append(腾讯控股(0700) PE: 18.2) if 百度 in query: results.append(百度(BIDU) PE: 11.8) if not results: return 未找到数据 return ; .join(results) tool def python_calculator(code: str): 用于计算。输入必须是 python 代码。 try: result eval(code) return f计算结果: {result} except Exception as e: return f计算错误: {e}这里的web_search使用的是虚假的模拟信息上面的工具描述不够完整但是能用如果用实际案例请描述完整工具的描述参考之前文章。初始化agent其他3个agentdef create_agent(state: dict, llm, tools, system_prompt): llm_tools llm.bind_tools(tools) prompt [SystemMessage(contentsystem_prompt)] state[messages] response llm_tools.invoke(prompt) results [response] for tool_call in response.tool_calls: func_name tool_call[name] args tool_call[args] call_id tool_call[id] func next((t for t in tools if t.name func_name), None) if func: tool_output func.invoke(args) tool_msg ToolMessage( contentstr(tool_output), namefunc_name, tool_call_idcall_id ) results.append(tool_msg) def researcher_node(state): return create_agent( statestate, llmllm, tools[web_search], system_prompt你是一个研究员。只负责查数据找到数据后直接输出原话不需要计算。 ) def coder_node(state): return create_agent( statestate, llmllm, tools[python_calculator], system_prompt你是一个程序员。根据上面研究员查到的数据写代码计算平均值。 ) def finish_node(state): return create_agent( statestate, llmllm, tools[], system_prompt任务完成简短的总结最终结果。 )流程相对简单create_agent细节前面文章已经讲解这里就不废话了。管理agentdef supervisor_node(state): system_prompt ( 你是项目经理。根据对话历史决定下一步交给谁。 查数据找 Researcher计算找 Coder识别图片找 Photographer。 如果任务完成必须选择 FINISH。 ) prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namemessages), (system, 根据以上情况请做出选择。), ]) class RouteResponse(BaseModel): next: Literal[Researcher, Coder, FINISH] Field( ..., description下一步交给谁如果任务完成请选 FINISH ) chain prompt | llm.with_structured_output(RouteResponse) response chain.invoke(state) return {next: response.next}这一步需要简单说明class RouteResponse(BaseModel): next: Literal[Researcher, Coder, FINISH] Field( ..., description下一步交给谁如果任务完成请选 FINISH ) chain prompt | llm.with_structured_output(RouteResponse)LLM中的with_structured_output方法是langchain提供的一个组件功能是限定LLM的输出格式返回相应格式的字典。定义输出格式限定的类该类是BaseModel的子类Literal是选择要求ai从Researcher, Coder, FINISH三选一Field描述变量尽量详尽描述例子因为是给大模型看的class classname(BaseModel): fieldname: fieldtype Field(..., description描述)构建图重要def init_agent(): workflow StateGraph(State) workflow.add_node(Researcher, researcher_node) workflow.add_node(Coder, coder_node) workflow.add_node(Supervisor, supervisor_node) workflow.add_node(Finish, finish_node) workflow.add_edge(Researcher, Supervisor) workflow.add_edge(Coder, Supervisor) workflow.add_edge(Finish, END) workflow.add_conditional_edges( Supervisor, lambda state: state[next],{ Researcher: Researcher, Coder: Coder, FINISH: Finish, }) workflow.set_entry_point(Supervisor) return workflow.compile()这一步相当于连接工作流构建的流程图如下--------------------------- | 开始 | -------------------------- | v --------------------------- | Supervisor |---------------- | (通过当前任务状态返回next) | | -------------------------- | |(根据state中next判断去向) | _____________|_____________ | / | \ | / | \ | v v v | ------------ ------------ ------------ | | Researcher | | Coder | | Finish | | | (Agent) | | (Agent) | | (Cleanup) | | ----------- ----------- ----------- | | | | | | | v | | | ------------ | --------------- | END | | | ------------ | | | -------------------------------------- (返回重新选择下一个agent)