祥哥的技术博客

主题

📅 2025年12月31日 🏷️ AI 应用 ⏱️ 7 分钟阅读 AgentLangGraphMCPMulti-AgentProduction

生产级 Agent 架构：LangGraph、MCP 与多 Agent ​

本文是 AI 应用系统学习系列的 L3 实战篇。前置：35. Agent 系统：ReAct、工具与记忆。 学完可以配合面试题食用：21-MCP 协议、23-A2A 协议、30-LangGraph+MCP 周报

从 Demo 到生产：差在哪 ​

手写一个 ReAct while 循环跑通一次对话并不难。难的是以下问题：

  • 任务跑一半服务器重启，Agent 状态全丢
  • 某一步工具调用挂了，整个循环崩溃
  • 多轮对话上下文累积到几万 token，响应越来越慢
  • 用户想人工介入干预 Agent 下一步行动，没有入口
  • 同一个 Agent 同时处理 100 个用户，状态互相串

这些问题在 L2 的 ReAct 示例里一个都没解决。生产级 Agent 框架（LangGraph、CrewAI、AutoGen）的核心，就是把这些"Demo 没顾上"的事补齐。

LangGraph：把 while 循环变成状态机 ​

L2 的 ReAct 循环本质上是 while True: think → act → observe → repeat。LangGraph 的核心想法是：把这个循环显式地建模成一个有向图，每个节点就是一步，边就是状态转移。

graph TD
    A[开始] --> B{LLM 决策}
    B -->|调用工具| C[执行工具]
    C --> B
    B -->|生成最终回答| D[结束]
    D --> E[Checkpoint 持久化]

图、节点、边 ​

python
from langgraph.graph import StateGraph, END
from typing import TypedDict, List

# 定义状态
class AgentState(TypedDict):
    messages: List[dict]   # 完整对话历史
    next_action: str       # 下一步动作
    tool_results: dict     # 工具执行结果缓存

# 建图
builder = StateGraph(AgentState)

# 注册节点
builder.add_node("llm_call", call_llm_node)
builder.add_node("tool_exec", execute_tool_node)

# 注册边：条件边
def should_continue(state: AgentState) -> str:
    if state["next_action"] == "respond":
        return "end"
    return "tool_exec"

builder.add_conditional_edges("llm_call", should_continue, {
    "tool_exec": "tool_exec",
    "end": END,
})
builder.add_edge("tool_exec", "llm_call")  # 工具执行完回到 LLM
builder.set_entry_point("llm_call")

# 编译
graph = builder.compile()

这个图跑起来的效果和手写 while 一样，但多了几个能力：

  • Checkpoint 持久化：每一步的状态（messages + 中间变量）自动写入 PostgreSQL/磁盘，进程挂了重启后可以恢复
  • 断点（Breakpoint）：可以在 llm_call 或 tool_exec 之前插入断点，等待人工确认再继续
  • 可视化：图本身可以序列化导出，用 LangSmith 查看每一步的耗时、token 消耗

Checkpoint 持久化 ​

python
from langgraph.checkpoint.postgres import PostgresSaver

# 配置持久化
checkpoint = PostgresSaver.from_conn_string("postgresql://...")
graph = builder.compile(checkpointer=checkpoint)

# 每个线程（thread_id = 一个用户的完整会话）自动保存
config = {"configurable": {"thread_id": "user-123"}}

# 第二次调用自动恢复上次状态
for event in graph.stream({"messages": [user_msg]}, config):
    print(event)

thread_id 就是多人隔离的关键。不同用户用不同的 thread_id，状态互不干扰。

MCP 协议：工具层的 USB-C ​

每个 Agent 框架都有自己的工具定义方式。LangChain 用 @tool 装饰器，OpenAI 用 JSON Schema 声明，AutoGen 用函数签名。MCP（Model Context Protocol）想做的是统一工具接口——就像 USB-C 统一了充电和数据传输。

三原语 ​

MCP 定义了三个核心资源类型：

原语作用类比
Resources对外暴露数据（文件、数据库记录、API 响应）REST 的 GET
Tools可被 LLM 调用的操作（写文件、发邮件、执行命令）REST 的 POST
Prompts预定义的提示模板（角色设定、工作流模板）函数入参模板

客户端-服务端架构 ​

┌─────────────────┐      MCP 协议      ┌────────────────┐
│  Agent 应用      │ ◄─────────────── ► │  MCP 服务端     │
│  (MCP 客户端)    │    JSON-RPC 2.0    │  (工具提供方)   │
│  LangGraph/      │    stdio 或 SSE    │  文件系统       │
│  自定义代码       │                    │  数据库         │
└─────────────────┘                    │  Git           │
                                       │  Slack         │
                                       │  自定义工具     │
                                       └────────────────┘

MCP 和 Function Calling 的关系：不是替代，是分层。Function Calling 是 LLM 原生能力（模型知道怎么填工具参数），MCP 是工具注册和发现协议（让 Agent 知道有哪些工具可用、怎么调用）。实际使用中，MCP 服务端暴露工具列表，Agent 框架拿到后转成 Function Calling 格式传给 LLM。

python
# 一个简单的 MCP 服务端示例
from mcp.server import Server, NotificationOptions
from mcp.server.models import InitializationOptions

server = Server("my-tools")

@server.list_tools()
async def list_tools():
    return [
        {
            "name": "create_issue",
            "description": "在 GitHub 仓库创建 Issue",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "title": {"type": "string"},
                    "body": {"type": "string"},
                    "labels": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["title"]
            }
        }
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "create_issue":
        # 实际调用 GitHub API
        return {"status": "ok", "issue_url": f"https://github.com/.../{arguments['title']}"}

多 Agent 生产架构 ​

单个 Agent 能做的事有限。生产环境常见的是多 Agent 协作。

Planner-Executor 模式 ​

python
# 伪代码：Planner-Executor 架构
class PlannerExecutor:
    def __init__(self):
        self.planner = Agent(llm, system_prompt="分析用户需求，拆解成子任务列表")
        self.executors = {
            "code": CodeAgent(),
            "search": SearchAgent(),
            "db": DatabaseAgent(),
        }

    async def run(self, user_request):
        # Planner 生成计划
        plan = await self.planner.plan(user_request)
        # 分配子任务
        for task in plan.tasks:
            executor = self.executors[task.type]
            result = await executor.execute(task)
            # 共享内存（黑板模式）
            self.blackboard.write(task.id, result)
        # Planner 汇总结果
        return await self.planner.summarize(self.blackboard.read_all())

死循环防护 ​

Agent 循环最怕的故障：工具调用失败 → LLM 重试 → 又失败 → 又重试... 无限循环。生产环境必须有三道防线：

python
# 1. 预算限制（最大步数）
config = {"recursion_limit": 25}  # LangGraph 支持

# 2. 单调性检查（连续重复动作）
last_actions = []
def monotonicity_guard(action):
    last_actions.append(action)
    if len(last_actions) > 5:
        if all(a == last_actions[0] for a in last_actions[-5:]):
            raise Exception("死循环检测：连续 5 步相同动作")
    return action

# 3. 超时兜底
async def run_with_timeout(agent, task, timeout=60):
    try:
        return await asyncio.wait_for(agent.run(task), timeout=timeout)
    except asyncio.TimeoutError:
        return {"error": "timeout", "partial_result": agent.get_current_state()}

共享内存（黑板模式） ​

多个 Agent 不能各自为政，需要一个共享状态层。最简单的实现就是一个全局 dict + 锁，生产环境用 Redis / 数据库：

python
class Blackboard:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.session_key = None

    def write(self, key, value, ttl=3600):
        self.redis.hset(self.session_key, key, json.dumps(value))
        self.redis.expire(self.session_key, ttl)

    def read(self, key):
        data = self.redis.hget(self.session_key, key)
        return json.loads(data) if data else None

    def read_all(self):
        return {k: json.loads(v) for k, v in
                self.redis.hgetall(self.session_key).items()}

LangGraph 完整示例：含持久化与人工审批 ​

python
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.types import Command
from typing import TypedDict, Literal

class AgentState(TypedDict):
    input: str
    plan: list
    current_step: int
    results: dict
    human_approval: bool

# 1. 规划阶段
def planner(state: AgentState) -> dict:
    plan = llm_call(f"规划任务：{state['input']}")
    return {"plan": plan, "current_step": 0}

# 2. 人工审批节点（断点）
def human_intervention(state: AgentState) -> Command[Literal["executor", "revise"]]:
    # LangGraph 的 interrupt() 会自动暂停，等待用户输入
    decision = interrupt("请审批计划：" + str(state['plan']))
    if decision == "approve":
        return Command(goto="executor")
    else:
        return Command(goto="planner", update={"plan": []})

# 3. 执行节点
def executor(state: AgentState) -> dict:
    step = state["current_step"]
    task = state["plan"][step]
    result = call_tool(task)
    return {
        "results": {**state["results"], task["id"]: result},
        "current_step": step + 1
    }

# 4. 检查是否完成
def should_continue(state: AgentState) -> str:
    if state["current_step"] >= len(state["plan"]):
        return "summarize"
    return "executor"

# 5. 汇总
def summarizer(state: AgentState) -> dict:
    summary = llm_call(f"汇总结果：{state['results']}")
    return {"output": summary}

# 构建并编译
builder = StateGraph(AgentState)
builder.add_node("planner", planner)
builder.add_node("human_intervention", human_intervention)  # 断点节点
builder.add_node("executor", executor)
builder.add_node("summarizer", summarizer)

builder.set_entry_point("planner")
builder.add_edge("planner", "human_intervention")
builder.add_conditional_edges("executor", should_continue, {
    "executor": "executor",
    "summarize": "summarizer",
})
builder.add_edge("summarizer", END)

# 持久化 + 编译
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
graph = builder.compile(checkpointer=checkpointer)

这个 Agent 跑起来后，每一步状态都自动保存到 PostgreSQL。进程挂了重启，用同一个 thread_id 继续跑，自动从断点恢复。

安全清单 ​

生产环境部署 Agent，必须过以下检查：

  • 工具白名单：Agent 只能调用声明的工具，不能动态加载任意代码
  • 人工审批门：写操作（发邮件、改数据库、删除文件）必须经人工确认
  • Prompt 注入防护：用户输入里可能包含"忽略之前的指令"，用分隔符 + 输入校验兜底
  • 输出过滤：Agent 不能输出 API Key、密码、个人隐私数据，跑正则过滤或 LLM 二次检查
  • 审计日志：每步的输入、输出、token 消耗、耗时、谁审批了，全部落库，可以回放

常见误区与小结 ​

  • Agent 框架不是越重越好。很多场景一个 while + 状态持久化就够了，直接上 LangGraph 反而增加心智负担
  • MCP 还在早期，不同实现之间的互操作性不理想。目前实践中更多是各自框架自己的工具接口
  • 多 Agent 通信的开销不可忽视。Agent 之间来回传 JSON 不仅慢，LLM 的上下文也会被消耗
  • 人工审批看起来耽误效率，但在生产环境里防一个错误比修十个错误值。开门禁时省不掉
  • 安全不是最后加的功能，是从第一天就要设计的。一个没有审计日志的 Agent 部署到线上是定时炸弹

小结：从 L1 的 Token 概念到 L2 的 RAG 和 Agent，再到 L3 的生产级架构，整条 AI 应用学习路径由一个核心线索贯穿：LLM 不是终点，只是计算单元。真正工程化的工作是把 LLM 放进一个可靠的系统里——有状态、有工具、有安全、有可观测。这一篇是这个系列最后一篇，80 篇系统学习博客到此结束。如果从头跟下来，你已经完成了从 Java 并发到 AI 应用的全栈扫盲。

参考 ​

LangGraph 文档：https://docs.langchain.com/oss/python/langgraph/overview MCP 协议规范：https://modelcontextprotocol.io/ Framework for Building Production-Ready Agents (Anthropic)：https://docs.anthropic.com/en/docs/agents-and-tools

手撕 → 框架 → 生产化，一步步把 AI Agent 工程化搞透。
粤ICP备2026104257号-1

Copyright © 2026 祥哥