模块化：独立的代理使开发、测试和维护代理系统变得更加容易。 专业化：可以创建专注于特定领域的专家代理，这有助于提高整体系统的性能。 控制：可以明确控制代理如何通信（而不是依赖于函数调用）。
在多代理系统中，有多种连接代理的方法：
- network：每个代理都可以与其它代理通信，任何代理都可以决定接下来要呼叫哪个代理。
- supervisor：每个代理与单个supervisor agent通信。主管代理决定下一步应呼叫哪个代理。
- supervisor（工具调用）：这是supervisor架构的一个特例。单个代理可以表示为工具。在这种情况下，主管代理使用工具调用 LLM 来决定调用哪些代理工具，以及要传递给这些代理的参数。
- 层次结构：我们可以定义一个具有多个supervisor的多代理系统。这是supervisor架构的泛化，允许更复杂的控制流。
- 自定义多代理工作流：每个代理仅与一部分代理进行通信。流程的某些部分是确定性的，只有部分代理可以决定接下来要调用哪些其他代理。
from langgraph.constants import ENDfrom typing_extensions import Literalfrom langchain_core.tools import toolfrom langgraph.graph import MessagesState, StateGraph, STARTfrom langgraph.types import Commandfrom langchain_core.messages import SystemMessage,ToolMessagefrom langchain_ollama import ChatOllamaimport base_confmodel = ChatOllama(base_url=base_conf.base_url,model=base_conf.model_name,temperature=0)# 下面的两个tool 不返回任何内容：# 我们只是用它作为一种提示：让LLM知道它需要交给另一个代理@tooldef transfer_to_multiplication_expert():"""请求乘法专家的帮助。"""return@tooldef transfer_to_addition_expert():"""请求加法专家的帮助。"""return# 下面的两个节点分别代表加法专家和乘法专家def addition_expert(state: MessagesState) -> Command[Literal["multiplication_expert", END]]:system_prompt = ("你是一个加法专家，你可以请求乘法专家帮助进行乘法。""在交接之前总是先完成你那部分的计算。")messages = [SystemMessage(content=system_prompt)] + state["messages"]ai_msg = model.bind_tools([transfer_to_multiplication_expert,transfer_to_addition_expert]).invoke(messages)# 首先先走到了加法这里，这里的ai_msg返回了一个tool_call# 如果有工具调用，LLM需要交给另一个代理if len(ai_msg.tool_calls) > 0:# 这里为什么官方教程总是用-1呢？因为这里的tool_call_id是一个list，所以-1是最后一个# 那有没有可能list中有多个呢？有可能，如果修改prompt，并且传入多个tool那么ai_msg是会返回多个tool_call的# 因此实际开发中，要注意这个问题tool_call_id = ai_msg.tool_calls[-1]["id"]# 注意：在这里插入ToolMessage很重要，因为LLM提供者期望所有AI message 后面都有相应的工具结果消息tool_msg = ToolMessage(content="成功转移", tool_call_id=tool_call_id)return Command(# 经过上面的处理，到达multiplication_expert的时候，入参已经变成了加法之后乘法需要的入参，# 也就是原来加法部分已经被计算了：8 * 12goto="multiplication_expert", update={"messages": [ai_msg, tool_msg]})# 如果专家有答案，直接返回给用户return Command(goto=END, update={"messages": [ai_msg]})def multiplication_expert(state: MessagesState,) -> Command[Literal["addition_expert", END]]:system_prompt = ("你是一个乘法专家，你可以请求加法专家帮助进行加法。""在交接之前总是先完成你那部分的计算。")messages = [SystemMessage(content=system_prompt)] + state["messages"]# 当这里拿着 8*12 的 AI message的时候，ai_msg最终的tool_call里面# 是没有transfer_to_multiplication_expert的# 它已经给出了结果，所以这里的tool_call是空的ai_msg = model.bind_tools([transfer_to_addition_expert]).invoke(messages)if len(ai_msg.tool_calls) > 0:tool_call_id = ai_msg.tool_calls[-1]["id"]tool_msg = ToolMessage(content="成功转移", tool_call_id=tool_call_id)return Command(goto="addition_expert", update={"messages": [ai_msg, tool_msg]})# 这里就直接返回了结果return Command(goto=END, update={"messages": [ai_msg]})builder = StateGraph(MessagesState)builder.add_node("addition_expert", addition_expert)builder.add_node("multiplication_expert", multiplication_expert)# 我们总是从加法专家开始builder.add_edge(START, "addition_expert")graph = builder.compile()from langchain_core.messages import convert_to_messagesdef pretty_print_messages(update):if isinstance(update, tuple):ns, update = update# 在打印输出中跳过父图更新if len(ns) == 0:returngraph_id = ns[-1].split(":")[0]print(f"来自子图 {graph_id} 的更新:")print("\n")for node_name, node_update in update.items():print(f"来自节点 {node_name} 的更新:")print("\n")for m in convert_to_messages(node_update["messages"]):m.pretty_print()print("\n")# 让我们运行图for chunk in graph.stream({"messages": [("user", "what's (3 + 5) * 12")]},):pretty_print_messages(chunk)
来自节点 addition_expert 的更新:================================== Ai Message ==================================Tool Calls:transfer_to_multiplication_expert (3e43e7c6-5ca4-433e-9a0e-e3a6f7913109)Call ID: 3e43e7c6-5ca4-433e-9a0e-e3a6f7913109Args:expression: 8 * 12================================= Tool Message =================================成功转移来自节点 multiplication_expert 的更新:================================== Ai Message ==================================The result of (3 + 5) * 12 is 96.
