2025年AI教学实战：用Python和LangGraph构建多智能体协作系统，从零实现复杂任务自动化编排

从单智能体到多智能体：2025年AI教学的核心范式跃迁

2025年，AI教学领域最显著的变化之一是智能体架构从单体走向协作。单一大模型在复杂任务处理中暴露出上下文窗口限制、领域知识冲突和推理路径不可控等问题。多智能体系统通过角色分工、任务分解和消息传递，为自动化编排复杂工作流提供了工程化解决方案。LangGraph作为LangChain生态中专门用于构建有状态、多轮对话和循环图执行框架的工具，正成为实现这一范式的核心基础设施。

本文将深入解析如何利用Python和LangGraph，构建一个具备任务规划、代码生成、代码审查与执行验证的多智能体协作系统。这套系统能够自动处理从需求分析到代码交付的完整闭环，适用于企业级自动化开发流水线。

LangGraph核心概念：状态图与条件边

传统LangChain的链式调用（Sequential Chain）无法处理需要循环和条件分支的场景。LangGraph通过引入状态图（StateGraph），允许开发者定义节点（Node）和边（Edge），其中边可以附加条件函数（Conditional Edge），从而构建非线性的执行流程。

核心数据结构是状态（State），它是一个可变的字典或Pydantic模型，在节点间传递。每个节点接收当前状态，执行操作后返回更新后的状态子集。这种设计天然支持日志记录、错误恢复和人工介入。

from langgraph.graph import StateGraph, END
from typing import TypedDict, List, Dict, Any

class AgentState(TypedDict):
    messages: List[Dict[str, str]]
    task: str
    generated_code: str
    review_result: str
    execution_output: str
    iteration_count: int

上述代码定义了一个自定义状态类型，包含任务描述、生成的代码、审查结果、执行输出和迭代计数器。这个状态将在整个多智能体系统中流转。

架构设计：四个智能体角色与协作机制

我们设计的多智能体系统包含四个核心角色，每个角色由独立的LLM调用和大模型提示词模板驱动：

• 规划智能体（Planner Agent）：解析用户输入的自然语言任务，将其拆解为可执行的子步骤，并生成初步的伪代码或技术方案。
• 编码智能体（Coder Agent）：基于规划方案，编写符合规范的Python代码。该智能体被配置为输出完整可运行代码块，并附带必要的注释。
• 审查智能体（Reviewer Agent）：对生成的代码进行静态分析，检查语法错误、潜在的性能瓶颈和安全漏洞。如果发现问题，返回包含具体修改建议的审查报告。
• 执行智能体（Executor Agent）：在沙箱环境中执行代码，捕获输出结果或异常信息。执行成功则返回输出，失败则返回错误栈。

协作机制采用循环反馈模式：编码智能体输出代码后，审查智能体介入。若审查通过，则转交执行智能体；若审查未通过，则将审查意见反馈给编码智能体进行修改。执行失败时，错误信息同样回流至编码智能体。系统设置最大迭代次数（例如3次），防止死循环。

实战搭建：用LangGraph定义节点与边

首先，定义各个智能体节点的实现函数。这些函数会调用底层的大语言模型（例如GPT-4o或Claude 3.5 Sonnet）。

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

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

def planner_node(state: AgentState) -> dict:
    system_prompt = "你是一位资深软件架构师。请分析用户任务，输出技术方案和伪代码。"
    response = llm.invoke([SystemMessage(content=system_prompt), HumanMessage(content=state["task"])])
    return {"messages": [response], "iteration_count": state.get("iteration_count", 0) + 1}

def coder_node(state: AgentState) -> dict:
    system_prompt = "你是一位高级Python工程师。请根据技术方案编写可运行的Python代码。只输出代码，不要多余解释。"
    task_context = state["messages"][-1].content if state["messages"] else state["task"]
    response = llm.invoke([SystemMessage(content=system_prompt), HumanMessage(content=task_context)])
    return {"generated_code": response.content}

审查节点和执行节点类似，但执行节点需要集成一个安全的代码执行环境。这里使用subprocess在临时目录中运行代码，并捕获输出。

import subprocess, tempfile, os

def executor_node(state: AgentState) -> dict:
    code = state["generated_code"]
    with tempfile.TemporaryDirectory() as tmpdirname:
        filepath = os.path.join(tmpdirname, "generated_script.py")
        with open(filepath, "w") as f:
            f.write(code)
        result = subprocess.run(["python", filepath], capture_output=True, text=True, timeout=30)
        output = result.stdout if result.returncode == 0 else result.stderr
    return {"execution_output": output}

关键步骤是定义条件边，决定流程走向。如果审查通过或迭代次数超限，则进入执行节点；否则返回编码节点。

def decide_next_node(state: AgentState) -> str:
    if state.get("iteration_count", 0) >= 3:
        return "executor"
    if "审查通过" in state.get("review_result", ""):
        return "executor"
    else:
        return "coder"

构建并编译状态图

将上述节点和边组装成可执行的图。

from langgraph.graph import StateGraph

workflow = StateGraph(AgentState)

workflow.add_node("planner", planner_node)
workflow.add_node("coder", coder_node)
workflow.add_node("reviewer", reviewer_node)
workflow.add_node("executor", executor_node)

workflow.set_entry_point("planner")
workflow.add_edge("planner", "coder")
workflow.add_edge("coder", "reviewer")
workflow.add_conditional_edges(
    "reviewer",
    decide_next_node,
    {
        "coder": "coder",
        "executor": "executor"
    }
)
workflow.add_conditional_edges(
    "executor",
    lambda state: "coder" if "Error" in state.get("execution_output", "") else END,
    {
        "coder": "coder",
        END: END
    }
)

app = workflow.compile()

编译后的app对象支持流式调用。通过app.stream()可以逐节点观察状态变化，便于调试和日志记录。

运行示例：自动生成并验证一个数据处理脚本

假设用户输入任务：“从CSV文件中读取销售数据，计算每个月的总销售额，并输出到新文件。”

initial_state = AgentState(
    messages=[],
    task="从CSV文件中读取销售数据，计算每个月的总销售额，并输出到新文件。",
    generated_code="",
    review_result="",
    execution_output="",
    iteration_count=0
)

for event in app.stream(initial_state):
    for node_name, state_update in event.items():
        print(f"--- 节点: {node_name} ---")
        if "generated_code" in state_update:
            print(f"代码生成: {state_update['generated_code'][:100]}...")
        if "execution_output" in state_update:
            print(f"执行输出: {state_update['execution_output'][:100]}...")

在测试中，系统第一次生成的代码可能未包含错误处理，审查智能体指出后，编码智能体在第二轮迭代中加入了try-except和文件存在性检查，最终执行成功。整个流程通过LangGraph的状态追踪机制清晰可见。

优化策略：提示词工程与记忆管理

多智能体系统的性能高度依赖提示词质量。对于审查智能体，建议采用结构化输出提示，强制其输出JSON格式的审查报告，包含passed布尔值和suggestions列表。编码智能体则应使用少样本示例，提供1-2个类似的代码片段作为参考。

记忆管理方面，LangGraph的状态本身承担了短期记忆功能。对于需要跨会话记忆的场景（例如用户偏好或项目上下文），可以将状态持久化到数据库（如Redis或PostgreSQL）。LangGraph提供了MemorySaver接口，可无缝集成。

from langgraph.checkpoint import MemorySaver

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

启用检查点后，系统可以支持人工中断和恢复（Human-in-the-loop），例如在代码审查节点之前暂停，等待人工确认。

部署与扩展：从原型到生产

将多智能体系统部署为REST API服务，推荐使用FastAPI。每个请求创建一个新的状态图实例，并通过唯一线程ID管理会话。

from fastapi import FastAPI
from pydantic import BaseModel

app_api = FastAPI()

class TaskRequest(BaseModel):
    task: str
    thread_id: str

@app_api.post("/run_agent")
async def run_agent(request: TaskRequest):
    initial_state = AgentState(messages=[], task=request.task, ...)
    config = {"configurable": {"thread_id": request.thread_id}}
    results = []
    for event in app.stream(initial_state, config):
        results.append(event)
    return {"status": "completed", "events": results}

扩展方向包括：增加RAG工具节点以访问外部知识库，集成代码格式化工具（如Black），以及对接CI/CD流水线。对于高并发场景，可结合Celery将任务异步化。

总结

2025年，多智能体系统不再是实验室概念，而是工程化解决复杂任务的标准范式。LangGraph通过有状态图执行框架，将传统链式调用无法处理的循环、分支和条件逻辑封装为可维护的代码结构。本文展示的四角色协作系统，从任务解析到代码交付，实现了端到端的自动化。开发者在此基础上，可根据业务需求灵活替换智能体角色、调整提示词策略或集成外部工具，构建真正可落地的AI驱动开发流水线。