一、 引言：单 Agent 的瓶颈与 Multi-Agent 的崛起

在构建复杂的大模型（LLM）应用时，单个 Agent（如传统的 ReAct 架构）往往会面临“能力过载”的问题。当一个 Agent 需要同时处理代码编写、数据分析、安全审计和日志输出等多项繁重任务时，其 Prompt 会变得异常臃肿，导致大模型在长文本中迷失方向，进而出现幻觉、工具误调用或逻辑死循环。

为了解决这一痛点，多智能体系统（Multi-Agent Systems, MAS） 成为了企业级 AIGC 落地的新范式。其核心思想类似于现代企业的组织架构：将一个复杂的巨型任务拆解，交由多个专注于特定领域的“专职 Agent”（如科研 Agent、编程 Agent、校对 Agent）共同协作完成。 通过明确的角色分工和严格的状态流转控制，多智能体系统能够显著提升复杂任务的成功率与稳定性。本文将基于 LangChain 生态中专为图结构设计的 LangGraph 框架，深入剖析多智能体协同网络的底层原理与代码实现。

二、 多智能体协同的核心拓扑结构

在设计 Multi-Agent 系统时，常见的协作模式主要有以下两种：

1. 协作网关模式（Hub and Spoke / Supervisor）

在这种模式下，存在一个“主管（Supervisor） Agent”，它负责接收用户输入，并根据任务性质将子任务分发给不同的“员工（Worker） Agent”。员工完成任务后将结果汇报给主管，由主管决定是继续分发还是结束任务。这种模式适合流程边界清晰、需要中心化调度的场景。

2. 流水线模式（Chains / Sequential Pipeline）

智能体之间像工业流水线一样串联，Agent A 的输出直接作为 Agent B 的输入。这种模式适合强顺序驱动的任务，例如：需求分析 -> 代码生成 -> 代码测试 -> 自动化部署。

本文将以 Supervisor（主管模式） 为例进行深度实战。

三、 基于 LangGraph 的多智能体实战准备

LangGraph 是 LangChain 团队推出的一个通过将步骤建模为图（Graph）中的顶点（Nodes）和边（Edges）来构建有状态、多角色应用的库。相比传统的链式结构，它完美支持循环（Cycles），这对于 Agent 的反思和迭代至关重要。

首先，安装核心依赖：

Bash

pip install langgraph langchain_openai langchain_core

配置核心 API 环境：

Python

import os

os.environ["OPENAI_API_KEY"] = "sk-xxxxxxxxxxxxxxxxxxxxxxxx"
os.environ["OPENAI_API_BASE"] = "https://api.openai.com/v1"

四、 核心代码实现：构建一个研发团队架构

我们将构建一个小型多智能体网络，包含三个角色：

  1. Coder（程序员）：负责根据需求编写 Python 代码。

  2. Tester（测试员）：负责运行并对代码进行质量检查。

  3. Supervisor（主管）：负责审阅全局，分发任务。

1. 定义全局状态（State）

在 LangGraph 中，所有 Agent 共享一个全局状态对象。各个节点通过向这个状态中追加或覆盖数据来进行通信。

Python

from typing import typing, Annotated, Sequence
from typing_extensions import TypedDict
import operator
from langchain_core.messages import BaseMessage

# 定义图的全局状态，messages 字段用于记录所有的对话历史
class TeamState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_agent: str  # 记录下一个执行任务的 Agent 名称

2. 创建独立智能体节点

接下来，我们定义 Coder 和 Tester 的逻辑。为了简化演示，我们使用 Prompt 为其注入不同的角色特性。

Python

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Coder 节点的生成函数
def coder_node(state: TeamState):
    messages = state["messages"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一名精通 Python 的高级软件工程师。请根据用户的需求编写高效、整洁的代码。只输出代码，不要多余的解释。"),
        MessagesPlaceholder(variable_name="messages"),
    ])
    chain = prompt | llm
    response = chain.invoke({"messages": messages})
    return {"messages": [response], "next_agent": "Supervisor"}

# Tester 节点的生成函数
def tester_node(state: TeamState):
    messages = state["messages"]
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一名严苛的测试工程师。请审阅团队生成的 Python 代码，指出其中潜在的 Bug、边界条件漏洞或性能问题。如果代码完美，请回复 'PASS'。"),
        MessagesPlaceholder(variable_name="messages"),
    ])
    chain = prompt | llm
    response = chain.invoke({"messages": messages})
    return {"messages": [response], "next_agent": "Supervisor"}

3. 构建 Supervisor（主管）决策大脑

Supervisor 需要决定下一步是让 Coder 修正代码、让 Tester 检查，还是认为任务已圆满结束（FINISH）。

Python

import json
from langchain_core.output_parsers import JsonOutputParser

def supervisor_node(state: TeamState):
    messages = state["messages"]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个研发团队的主管。你的职责是根据当前团队的对话进展，决定下一步由谁来工作。可以选择的角色有：'Coder'、'Tester'。如果测试已经通过（收到PASS）或者任务已经圆满完成，请选择 'FINISH'。请务必返回 JSON 格式，例如：{{\"next\": \"Coder\"}}"),
        MessagesPlaceholder(variable_name="messages"),
    ])
    
    chain = prompt | llm
    response = chain.invoke({"messages": messages})
    
    # 解析模型输出的路由决定
    try:
        route_decision = json.loads(response.content)
        next_step = route_decision.get("next", "FINISH")
    except:
        next_step = "FINISH"
        
    return {"next_agent": next_step}

4. 使用 LangGraph 组装有向工作流图

现在，我们将上述节点和条件路由组合成一个完整的有向图。

Python

from langgraph.graph import StateGraph, END

# 1. 初始化有状态的图
workflow = StateGraph(TeamState)

# 2. 将编写好的节点加入图中
workflow.add_node("Coder", coder_node)
workflow.add_node("Tester", tester_node)
workflow.add_node("Supervisor", supervisor_node)

# 3. 配置条件边：Supervisor 根据 next_agent 的值进行分流
def router(state: TeamState):
    next_agent = state["next_agent"]
    if next_agent == "FINISH":
        return END
    return next_agent

# 从 Supervisor 出发，根据 router 函数的返回值跳转到 Coder、Tester 或结束
workflow.add_conditional_edges(
    "Supervisor",
    router,
    {
        "Coder": "Coder",
        "Tester": "Tester",
        "END": END
    }
)

# 4. 固定边：Worker 执行完后无条件返回给 Supervisor
workflow.add_edge("Coder", "Supervisor")
workflow.add_edge("Tester", "Supervisor")

# 5. 设置图的起点
workflow.set_entry_point("Supervisor")

# 6. 编译图，得到最终的可执行应用
app = workflow.compile()

五、 系统运行与协作链路解析

我们向这个编译好的多智能体网络投递一个开发需求：

Python

from langchain_core.messages import HumanMessage

inputs = {
    "messages": [HumanMessage(content="帮我写一个 Python 函数，实现快速排序算法，并包含简单的单元测试。")]
}

for output in app.stream(inputs):
    for key, value in output.items():
        print(f"--- 节点 [{key}] 执行完毕 ---")
        if "next_agent" in value:
            print(f"流转状态 -> 下一步指派给: {value['next_agent']}\n")

当程序启动后，你会观察到以下严密的团队协作流程：

  1. Supervisor 接收到原始需求，识别出这是一个编码任务，将状态置为 Coder。

  2. Coder 节点介入，生成标准快速排序代码，并自动将结果推回给 Supervisor。

  3. Supervisor 审查当前最新的代码消息，认为需要质量保障，于是将状态修改为 Tester。

  4. Tester 节点启动，对 Coder 生成的代码进行边界条件核查，判定逻辑无误，输出 PASS。

  5. Supervisor 捕捉到 PASS 信号，认为项目符合交付标准，将路由导向 END。

整个过程中，Agent 之间各司其职，通过修改和读取全局的 TeamState 实现了信息的无缝传递和状态的严密流转。

六、 总结：生产环境下的多智能体设计原则

在实际的企业级复杂业务场景中，构建 Multi-Agent 系统通常需要遵循以下设计原则：

  1. 原子化职责： 每一个子 Agent 的功能应当尽可能单一且聚焦。如果一个 Agent 既要负责查数据库，又要负责生成前端 UI，那就应当将其拆分为两个更小颗粒度的智能体。

  2. 状态裁剪： 在大型长对话中，不要把所有的原始日志都扔进全局 State 中。建议每个 Agent 在向全局状态汇报时，只提取关键上下文（摘要），以避免 Token 浪费和信息冗余导致的模型长文本迷失。

  3. 动态与静态结合： 虽然大模型具备动态路由的能力，但对于极其严谨的业务流（如财务审批），应当在 LangGraph 中使用硬编码边（Explicit Edges）来强制约束流转路径，而将大模型的推理能力留在节点内部，实现“宏观可控，微观灵活”。

# 网络
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

AI Agent 开发实战（01）：开发环境搭建与 Claude Code 安装

avatar AI Agent技术社区
cover

AI Agent_项目技术选型指南：从框架到落地的完整决策树

avatar AI Agent技术社区
cover

跨马翻译：批量图片视频智能处理，一站式跨境电商翻译解决方案

avatar AI Agent技术社区