LangGraph多代理协作系统构建与实战-AI智能范式网

LangGraph多代理协作系统构建与实战

李在田·2026-09-04·阅读 22 GEO检测（免费）

LangGraph多代理系统AI协作框架

|难度：普通|阅读约 18 分钟

1. LangGraph多代理协作实战指南

在AI应用开发中，单个代理处理复杂任务时往往会遇到瓶颈。今天我将分享如何使用LangGraph构建多代理协作系统，通过"分而治之"的策略让专业代理各司其职，共同完成复杂任务。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. 多代理协作的核心设计

2.1 问题背景与解决方案

当面对需要多种工具和领域知识的复杂任务时，单个代理（即使是GPT-4这样的强大模型）往往表现不佳。这就像让一位全科医生同时处理外科手术、眼科检查和儿科诊疗——虽然可能完成，但效率和质量都难以保证。

解决方案是创建多个专业代理：

研究代理（Researcher）：负责数据收集和验证
图表生成代理（Chart Generator）：负责数据可视化和代码执行
每个代理专注于自己的专业领域，通过协作完成整体任务

2.2 系统架构设计

整个系统的工作流程如下：

用户提出问题
研究代理进行数据收集
图表生成代理处理数据可视化
代理间循环协作直到问题解决
输出最终结果

这种架构的优势在于：

每个代理只需掌握特定工具，降低复杂度
错误更容易定位和修复
系统可扩展性强，可随时添加新代理

3. 实现步骤详解

3.1 环境准备与依赖安装

首先需要安装必要的Python包：

bash复制pip install -U langchain langchain_openai langsmith pandas langchain_experimental matplotlib langgraph langchain_core

然后设置API密钥（以Tavily搜索为例）：

python复制import os
os.environ["TAVILY_API_KEY"] = "your_api_key_here"

3.2 代理创建与配置

3.2.1 创建基础代理

我们定义一个通用的代理创建函数：

python复制from langchain_core.messages import BaseMessage, HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

def create_agent(llm, tools, system_message: str):
    prompt = ChatPromptTemplate.from_messages([
        ("system", 
         "你是一个有帮助的AI助手，与其他助手合作。"
         "使用提供的工具来推进问题的回答。"
         "如果你不能完全回答，没关系，另一个拥有不同工具的助手"
         "会接着你的位置继续帮助。执行你能做的以取得进展。"
         "如果你或其他助手有最终答案或交付物，"
         "在你的回答前加上FINAL ANSWER，以便团队知道停止。"
         "你可以使用以下工具: {tool_names}。\n{system_message}"),
        MessagesPlaceholder(variable_name="messages")
    ])
    return prompt.partial(
        system_message=system_message,
        tool_names=", ".join([tool.name for tool in tools])
    ) | llm.bind_tools(tools)

3.2.2 初始化专业代理

创建两个专业代理：

python复制from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")

# 研究代理 - 负责数据收集
research_agent = create_agent(
    llm,
    [tavily_tool],  # 搜索工具
    system_message="你应该提供准确的数据供chart_generator使用。"
)

# 图表生成代理 - 负责数据可视化
chart_agent = create_agent(
    llm,
    [python_repl],  # Python执行工具
    system_message="你展示的任何图表都将对用户可见。"
)

3.3 工具定义与集成

3.3.1 搜索工具配置

使用Tavily进行网络搜索：

python复制from langchain_community.tools.tavily_search import TavilySearchResults
tavily_tool = TavilySearchResults(max_results=5)

3.3.2 Python执行工具

创建安全的Python执行环境：

python复制from langchain_experimental.utilities import PythonREPL
from langchain_core.tools import tool
from typing import Annotated

repl = PythonREPL()

@tool
def python_repl(
    code: Annotated[str, "要执行以生成图表的Python代码。"],
):
    """执行Python代码并返回结果"""
    try:
        result = repl.run(code)
    except Exception as e:
        return f"执行失败: {repr(e)}"
    return f"执行成功:\n```python\n{code}\n```\n输出: {result}"

注意：PythonREPL在本地执行代码，存在安全风险。在生产环境中应考虑使用沙箱环境或更严格的安全措施。

3.4 状态图设计与实现

3.4.1 定义状态结构

python复制import operator
from typing import Annotated, Sequence, TypedDict

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    sender: str

3.4.2 创建代理节点

python复制import functools
from langchain_core.messages import AIMessage

def agent_node(state, agent, name):
    result = agent.invoke(state)
    if not isinstance(result, ToolMessage):
        result = AIMessage(**result.dict(exclude={"type", "name"}), name=name)
    return {
        "messages": [result],
        "sender": name
    }

research_node = functools.partial(agent_node, agent=research_agent, name="Researcher")
chart_node = functools.partial(agent_node, agent=chart_agent, name="chart_generator")

3.4.3 路由逻辑实现

python复制from typing import Literal

def router(state) -> Literal["call_tool", "__end__", "continue"]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "call_tool"
    if "FINAL ANSWER" in last_message.content:
        return "__end__"
    return "continue"

3.4.4 构建完整工作流

python复制from langgraph.graph import StateGraph, END, START
from langgraph.prebuilt import ToolNode

workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("Researcher", research_node)
workflow.add_node("chart_generator", chart_node)
workflow.add_node("call_tool", ToolNode([tavily_tool, python_repl]))

# 设置边和条件
workflow.add_conditional_edges(
    "Researcher",
    router,
    {"continue": "chart_generator", "call_tool": "call_tool", "__end__": END}
)

workflow.add_conditional_edges(
    "chart_generator",
    router,
    {"continue": "Researcher", "call_tool": "call_tool", "__end__": END}
)

workflow.add_conditional_edges(
    "call_tool",
    lambda x: x["sender"],
    {"Researcher": "Researcher", "chart_generator": "chart_generator"}
)

workflow.add_edge(START, "Researcher")
graph = workflow.compile()

4. 系统运行与测试

4.1 执行查询任务

让我们测试系统获取AI市场规模数据并生成图表的能力：

python复制events = graph.stream(
    {
        "messages": [
            HumanMessage(
                content="获取过去5年AI软件市场规模，"
                        "然后绘制一条折线图。"
                        "一旦你编写好代码，完成任务。"
            )
        ],
    },
    {"recursion_limit": 150},
)

for s in events:
    print(s)
    print("----")

4.2 典型执行流程分析

研究代理接收用户问题
使用Tavily搜索相关数据
将搜索结果传递给图表生成代理
图表代理生成Python代码绘制图表
系统检查结果质量，必要时循环改进
输出最终结果

4.3 常见问题与调试技巧

问题1：工具调用失败

检查工具是否正确定义和绑定
验证API密钥和网络连接
查看代理的提示词是否明确工具使用方法

问题2：代理间通信不畅

确保状态结构正确包含所有必要信息
检查路由逻辑是否能正确处理各种消息类型
验证每个代理的输出格式是否符合预期

问题3：无限循环

设置合理的recursion_limit
确保代理能在适当时候发出FINAL ANSWER
添加超时机制作为额外保护

5. 高级应用与优化建议

5.1 性能优化技巧

缓存机制：对频繁查询的数据实现缓存
并行处理：允许某些代理并行工作
结果验证：添加专门的结果验证代理
错误恢复：实现更健壮的错误处理流程

5.2 扩展更多专业代理

系统可以轻松扩展更多专业代理：

数据清洗代理：处理原始数据
报告生成代理：整理最终报告
质量检查代理：验证结果准确性

添加新代理只需：

定义代理的专业领域和工具
创建对应的节点
更新路由逻辑

5.3 监控与日志记录

实现完善的监控系统：

python复制from langsmith import Client

client = Client()

def log_to_langsmith(state):
    client.create_run(
        inputs={"state": str(state)},
        outputs={},
        run_type="log"
    )
    return state

workflow.add_node("monitor", log_to_langsmith)
workflow.add_edge("call_tool", "monitor")

6. 安全与最佳实践

代码执行安全：
- 使用沙箱环境执行Python代码
- 限制可导入的模块
- 设置资源使用限制
数据隐私：
- 匿名化处理敏感数据
- 遵守数据保护法规
- 谨慎选择第三方API
系统稳定性：
- 实现超时机制
- 添加熔断保护
- 定期测试故障恢复

在实际项目中，我建议先从简单任务开始，逐步增加复杂性。每次添加新代理或工具时，都要进行充分的测试。记录下代理间的典型对话模式，这能帮助优化路由逻辑和提示词设计。