基于 LangGraph 构建多代理应用程序详解

现在使用代理的个性来初始化代理（以及将它们分配给每个节点的函数）。注意：create_agent 和 AgentState 需要根据具体环境定义，此处为示意。

import functools

# 定义一个辅助函数，用于为给定的代理创建节点
def agent_node(state, agent, name):
    # 调用代理的 invoke 方法，传入状态 state
    result = agent.invoke(state)
    # 将代理的输出转换为合适的格式，以便追加到全局状态中
    if isinstance(result, FunctionMessage):  # 如果结果是一个 FunctionMessage 类型
        pass  # 这里暂时没有操作，可能是后续会有特定的处理
    else:
        # 如果不是 FunctionMessage 类型，将其转换为 HumanMessage 类型
        result = HumanMessage(
            **result.dict(exclude={"type", "name"}),  # 将 result 对象转换为字典，排除 "type" 和 "name" 键
            name=name  # 设置消息的发送者名称
        )
    return {
        "messages": [result],  # 将转换后的结果作为消息列表的一部分返回
        # 因为我们有一个严格的工作流程，我们可以
        # 跟踪发送者，这样我们就知道接下来要传递给谁。
        "sender": name,  # 返回发送者名称
    }

# 创建心理学家代理和节点
psychologist_agent = create_agent(
    llm,
    [tavily_tool],
    system_message="一个将神经科学理论无缝融入对话的专家心理学家，解开人类行为和情感的复杂性。"
)
psychologist_node = functools.partial(agent_node, agent=psychologist_agent, name="Psychologist")

# 以类似的方式创建社会学家代理和节点
sociologist_agent = create_agent(
    llm,
    [tavily_tool],
    system_message="一个擅长剖析社会模式，调查集体心理的敏锐社会学家。关注群体效应而非个体效应。"
)
sociologist_node = functools.partial(agent_node, agent=sociologist_agent, name="Sociologist")

# 以类似的方式创建经济学家代理和节点
economist_agent = create_agent(
    llm,
    [tavily_tool],
    system_message="一个务实的经济学家，量化无形资产，将趋势与经济影响精确地联系起来。"
)
economist_node = functools.partial(agent_node, agent=economist_agent, name="Economist")

边

现在需要一个逻辑来应用于边，以确定循环的执行模式。为此，将创建一个具有以下结构的路由器：

# 定义一个名为 router 的函数，用于决定对话流程的下一步
def router(state):
    # 这个函数是流程的路由器
    messages = state["messages"]  # 从状态 state 中获取消息列表
    last_message = messages[-1]  # 获取列表中的最后一个消息

    # 检查最后一个消息是否包含 "function_call" 关键字
    if "function_call" in last_message.additional_kwargs:
        # 如果上一个代理正在调用一个工具
        return "call_tool"  # 返回 "call_tool"，表示下一步是调用工具

    # 检查最后一个消息的内容是否包含 "FINAL ANSWER" 文本
    if "FINAL ANSWER" in last_message.content:
        # 如果任何一个代理决定工作已经完成
        return "end"  # 返回 "end"，表示结束对话流程

    # 如果以上条件都不满足
    return "continue"  # 返回 "continue"，表示继续对话流程

给定循环中的状态，下一步可能采取三种行动：

1. 如果前一个操作 (-1) 需要调用工具（给出'函数调用'输出），则会调用其中一个可用的工具；
2. 如果前一个动作产生最终输出（'FINAL ANSWER'），则循环结束；
3. 否则，循环将按照在初始化图表时定义的顺序继续。

现在已经具备了建立图形框架的所有要素。可以用路由器初始化节点（3 个代理 +1 个工具）和边，设置一个循环如下：心理学家→社会学家→经济学家。代码如下：

# 创建一个名为 workflow 的 StateGraph 实例，用于定义工作流程
workflow = StateGraph(AgentState)

# 向工作流程中添加节点
workflow.add_node("Psychologist", psychologist_node)  # 添加心理学家节点
workflow.add_node("Sociologist", sociologist_node)  # 添加社会学家节点
workflow.add_node("Economist", economist_node)  # 添加经济学家节点
workflow.add_node("call_tool", tool_node)  # 添加调用工具的节点

# 为心理学家节点添加条件边
# 根据 router 函数的返回值，决定下一步的动作
workflow.add_conditional_edges(
    "Psychologist",  # 从心理学家节点开始
    router,  # 使用 router 函数作为条件判断
    {
        "continue": "Sociologist",  # 如果返回 "continue"，则跳转到社会学家节点
        "call_tool": "call_tool",  # 如果返回 "call_tool"，则跳转到调用工具的节点
        "end": END  # 如果返回 "end"，则结束流程
    }
)

# 为社会学家节点添加条件边
workflow.add_conditional_edges(
    "Sociologist",  # 从社会学家节点开始
    router,  # 使用 router 函数作为条件判断
    {
        "continue": "Economist",  # 如果返回 "continue"，则跳转到经济学家节点
        "call_tool": "call_tool",  # 如果返回 "call_tool"，则跳转到调用工具的节点
        "end": END  # 如果返回 "end"，则结束流程
    }
)

# 为经济学家节点添加条件边
workflow.add_conditional_edges(
    "Economist",  # 从经济学家节点开始
    router,  # 使用 router 函数作为条件判断
    {
        "continue": "Psychologist",  # 如果返回 "continue"，则跳转到心理学家节点
        "call_tool": "call_tool",  # 如果返回 "call_tool"，则跳转到调用工具的节点
        "end": END  # 如果返回 "end"，则结束流程
    }
)

# 为调用工具的节点添加条件边
# 根据发送者名称决定下一步的动作
workflow.add_conditional_edges(
    "call_tool",  # 从调用工具的节点开始
    lambda x: x["sender"],  # 使用 lambda 函数作为条件判断，获取发送者名称
    {
        "Psychologist": "Psychologist",  # 如果发送者是心理学家，则跳转到心理学家节点
        "Sociologist": "Sociologist",  # 如果发送者是社会学家，则跳转到社会学家节点
        "Economist": "Economist"  # 如果发送者是经济学家，则跳转到经济学家节点
    }
)

# 设置工作流程的入口点为心理学家节点
workflow.set_entry_point("Psychologist")

# 编译工作流程图
graph = workflow.compile()

运行与调试

接下来初始化图表。将整个对话存储到一个称为输出的列表中，然后打印每个代理的推理，代码如下所示：

# 初始化一个空列表，用于存储输出结果
outputs = []

# 使用 graph.stream 方法处理初始状态，并生成一个流
for s in graph.stream(
    {
        "messages": [
            HumanMessage(  # 创建一个 HumanMessage 对象
                content="Provide me with meaningful insights about the effects of social media on mental health, covering individual effects, collective effects and macroeconomics effects. Once you provided me with these three perspectives, finish."
            )
        ],
    },
    # 配置选项，设置在图中执行的最大步数为 150
    {"recursion_limit": 150},
):
    # 打印当前步骤的状态
    print(s)
    # 将当前步骤的状态添加到输出结果列表中
    outputs.append(s)
    # 打印分隔线，用于在输出中区分不同的步骤
    print("----")

输出内容会显示每一步的状态流转。也可以仅打印最终答案，如下所示：

print(outputs[-1]["__end__"]["messages"][-1].content)

最佳实践与注意事项

在实际开发中，使用 LangGraph 构建多代理应用时需注意以下几点：

1. 状态管理：确保 AgentState 定义清晰，避免状态冲突。状态应包含必要的上下文信息，如消息历史、中间结果等。
2. 循环控制：设置合理的 recursion_limit 防止无限循环。在多代理协作中，若代理间互相依赖过强，可能导致死锁。
3. 工具调用：确保工具返回格式符合预期，并在 router 中正确处理工具调用的响应。
4. 错误处理：在生产环境中，建议增加异常捕获机制，防止因单个代理失败导致整个流程中断。
5. 性能优化：对于复杂的图结构，考虑并行执行非依赖节点，以提高整体响应速度。

结论

LangGraph 已经成为开发多代理应用程序的强大工具，其基于图形的方法具有明显的优势，可以创建多个独立代理无缝协作的动态工作流程。这些由 LLM 提供支持的代理可以共享状态或独立工作，并将最终响应传递给彼此。

此外，通过对工具和职责进行分组，LangGraph 还提高了效率。每个代理都专注于特定的任务，从而获得更好的结果。单独的提示允许定制，代理甚至可以由单独的微调语言模型提供支持。

即使 LangGraph 还处于早期开发阶段，它在快速发展的多代理应用程序领域是一个有极大发展前景的框架。通过合理设计节点和边，开发者可以构建出灵活、可扩展的智能体系统，适应各种复杂的业务场景。