用 LangChain 构建多 Agent AI 系统

想象你构建了一个能搜索网页的 AI Agent，又构建了一个能读取文档的 Agent，还有第三个能写代码的 Agent。现在，如果能让它们相互对话、辩论观点、传递任务，直到解决单个 Agent 无法独自处理的问题——这就是多 Agent AI 系统的威力。借助 LangChain，你今天就可以构建这样一个系统，无需分布式系统的博士学位。

多 Agent 系统并非只是 buzzword——它们是 AI 应用的下一个演进方向。与其依赖一个单体模型完成所有工作，不如把任务分解为专业角色：研究员、评论员、作家，甚至可能还有一个代码执行器。每个 Agent 专注于自己的工作、与他人共享上下文、迭代优化，直到输出质量过硬。最棒的是？LangChain 让这一切出人意料地触手可及。

让我们从头构建一个真实可运行的多 Agent 系统，你可以在笔记本上用不到 10 分钟跑起来。

为什么多 Agent 系统至关重要

当任务变得复杂时，单 Agent AI 常常会遇到瓶颈。它可能产生事实性幻觉、遗漏上下文，或无法验证自己的输出。多 Agent 系统通过引入专业化、协作和迭代来解决这些问题。

根据生产环境指南，多 Agent 系统的关键成功因素包括：

• Agent 专业化：每个 Agent 都有清晰、专注的职责 [2]。
• 通信模式：定义 Agent 之间如何交换信息 [2]。
• 状态管理：在交互过程中维护一致的状态 [2]。
• 错误处理：构建能够从故障中恢复的弹性系统 [2]。
• 监控：使用 LangSmith 等工具追踪性能 [2]。

在实践中，这意味着你的"研究员"Agent 永远不会尝试写散文，你的"评论员"Agent 永远不会去搜索网页，而你的"作家"Agent 只润色评论员验证过的内容。

架构：Supervisor + Sub-Agents

对初学者最有效的模式是 supervisor 模型：一个中央 Agent（supervisor）决定由哪个 sub-agent 处理下一步、分发任务并聚合最终结果。

LangChain 的 LangGraph 库让这件事变得简单。你需要定义：

• Sub-agents（如 search、read、write）
• 每个 Agent 可以使用的工具
• 一个编排流程的 supervisor [6][8]

这种方法让你的系统保持简单、可测试和可扩展。

构建你的第一个多 Agent 系统

让我们创建一个包含三个 Agent 的系统：

• Search Agent：在网上查找相关信息
• Reader Agent：从 URL 中提取关键洞察
• Writer Agent：将发现综合成一份清晰的报告

为简单起见，我们将使用 OpenAI 的 gpt-4o（或任何兼容模型）和 LangChain 的 create_react_agent。

Step 1: 安装依赖

pip install langchain langchain-community langgraph openai

Step 2: 设置环境

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchResults, WikipediaQueryRun
from langchain_community.utilities import WikipediaAPIWrapper
from langgraph import StateGraph, Command
from typing import TypedDict, List

# Initialize the model
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# Define tools
search_tool = DuckDuckGoSearchResults()
wiki_tool = WikipediaQueryRun(api_wrapper=WikipediaAPIWrapper())
tools = [search_tool, wiki_tool]

Step 3: 定义 Agent 状态

class AgentState(TypedDict):
    task: str
    search_results: List[str]
    wiki_content: str
    draft: str
    final_report: str

Step 4: 创建 Sub-Agents

def search_agent(state: AgentState):
    results = search_tool.invoke({"query": state["task"]})
    return {"search_results": [r["snippet"] for r in results]}

def reader_agent(state: AgentState):
    # Pick the first URL from search results (simplified)
    if not state["search_results"]:
        return {"wiki_content": ""}
    url = "https://en.wikipedia.org/wiki/" + state["task"].split()[0]  # Simplified
    content = wiki_tool.invoke({"query": state["task"].split()[0]})
    return {"wiki_content": content}

def writer_agent(state: AgentState):
    prompt = f"""
    Based on the following search results and Wikipedia content, write a concise report on: {state['task']}

    Search Results:
    {state['search_results']}

    Wikipedia Content:
    {state['wiki_content']}

    Write a clear, well-structured report.
    """
    response = llm.invoke(prompt)
    return {"draft": response.content}

Step 5: 构建 Supervisor 图

def supervisor(state: AgentState):
    if not state["search_results"]:
        return Command(update=search_agent(state), goto="search_agent")
    elif not state["wiki_content"]:
        return Command(update=reader_agent(state), goto="reader_agent")
    elif not state["draft"]:
        return Command(update=writer_agent(state), goto="writer_agent")
    else:
        return Command(goto="__end__")

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("search_agent", search_agent)
workflow.add_node("reader_agent", reader_agent)
workflow.add_node("writer_agent", writer_agent)
workflow.add_node("supervisor", supervisor)

workflow.set_entry_point("supervisor")
workflow.add_edge("supervisor", "supervisor")  # Loop until done

app = workflow.compile()

initial_state = {"task": "climate change impacts on coastal cities", 
                 "search_results": [], 
                 "wiki_content": "", 
                 "draft": "", 
                 "final_report": ""}

result = app.invoke(initial_state)
print(result["draft"])

执行流程如下：

1. 搜索"气候变化对沿海城市的影响"
2. 获取关于首要主题的 Wikipedia 内容
3. 综合两个来源写一份简洁的报告

你今天就可以用自己的 API key 运行这个。只需在你的环境中替换 OPENAI_API_KEY 即可。

让它成为生产级别

上面的例子是一个最小原型。要扩展它，需要：

• 添加更多 Agent：例如一个验证报告的"评论员" [2][4]
• 使用 LangSmith：追踪执行流程并监控延迟 [2]
• 实现错误处理：重试失败的工具调用，回退到备选工具 [2]
• 持久化状态：将中间结果保存到数据库以便审计 [2]

LangGraph 的 Command 路由允许你根据 supervisor 的决策动态切换 Agent，使你的系统灵活而富有弹性 [6]。

多 Agent AI 不是遥远的未来——它现在就已经可用了，而 LangChain 让任何水平的开发者都能上手。你不需要重建整个技术栈；只需从两三个专业 Agent 开始，分别测试它们，然后用 supervisor 将它们连接起来 [2]。

你的第一个系统可能很简单：一个查找数据的研究员、一个格式化数据的作家、一个检查准确性的评论员。但一旦你看到它们如何协作，你就会意识到——这才是 AI 规模化扩展的方式。

1. 把上面的代码复制到 multi_agent.py 文件中
2. 设置你的 OPENAI_API_KEY
3. 运行 python multi_agent.py，观察你的 Agent 协作

在这个 Dev.to 上分享你的结果并 @ 我——我很想看看你构建了什么！

AI 的未来不是一个巨大的模型。它是一组智能、专业化的 Agent 协同工作。而你现在已经准备好构建这样的团队了。

如果你觉得这篇文章有帮助，考虑请我喝杯咖啡 ☕ ——这让我能持续输出这些文章！

也可以看看我的 AI 工具集合：AI 次元世界 — 开发者的免费 AI 工具。