LangGraph实战：让Agent从脚本变成生产级可控系统

晨xi的光

312人浏览 · 2026-08-02 14:26:48

晨xi的光 · 2026-08-02 14:26:48 发布

聊《LangGraph真能提效吗？先看流程里最慢的那一步》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。

摘要

上周业务方提了一个需求：让客服Agent能自主查询订单、调用退款接口，但必须在关键步骤人工确认。我接手后才发现，之前写的"能跑通"的Agent代码，离生产环境差得远。权限怎么隔离？日志怎么追踪？出问题怎么回滚？用LangGraph重写后，这些问题有了标准答案。本文复盘这次改造过程，重点讲清楚State设计、条件分支、人工审批节点，以及工程化落地的几个关键判断。

---

为什么你的Agent能跑Demo却不敢上线

文章插图 1

之前写过不少Agent Demo，调用Llama API、拼接prompt、返回结果，跑起来都很丝滑。但一旦要上线，问题就来了：

退款接口不能随便调，谁来授权？
查询订单失败了，日志在哪里？
流程跑到一半崩了，怎么恢复？
业务方说"这个步骤要人工确认"，代码里怎么体现？

这些问题的共同点：Demo阶段不需要考虑，但生产环境必须解决。很多人用链式调用写Agent，代码越来越长，改不动、测不了、排错难。LangGraph的核心价值，是把"脚本"变成"系统"——流程可描述、状态可追踪、节点可干预。

---

State设计：把状态显式化

文章插图 2

写Agent最容易踩的坑：状态散落在各处，函数调用靠隐式传递。LangGraph要求你把State定义清楚，每个Node只操作State，不依赖外部变量。

from typing import TypedDict, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    # 用户输入
    user_query: str
    # 中间结果
    order_id: Annotated[str, operator.add]
    refund_amount: float
    # 决策结果
    decision: str  # "approve" / "reject" / "pending"
    # 审批记录
    approval_log: Annotated[list, operator.add]
    # 最终输出
    response: str

注意这里的operator.add，它表示这个字段是累加型的——每次写入都会追加，而不是覆盖。这对于日志、审批记录这类字段非常有用。

判断标准：如果你的State里有字典嵌套、或者Node之间传递临时变量，说明设计有问题。State应该是扁平的、可序列化的、能反映完整流程的。

---

Node与Edge：让流程可观测

Demo阶段，一个函数搞定所有逻辑。生产环境，每个Node应该是独立的、可测试的、可插拔的。

def query_order(state: AgentState) -> AgentState:
    """查询订单节点"""
    query = state["user_query"]
    # 调用订单服务，实际项目中这里应该有重试和超时控制
    order = order_service.search(query)
    state["order_id"] = order.id
    state["refund_amount"] = order.total
    return state

def check_permission(state: AgentState) -> AgentState:
    """权限校验节点"""
    user_role = get_current_user_role()
    if user_role not in ["admin", "refund_operator"]:
        state["decision"] = "reject"
        state["approval_log"].append(f"{datetime.now()}: 权限不足，拒绝退款请求")
    else:
        state["decision"] = "pending"
    return state

def ask_human(state: AgentState) -> AgentState:
    """人工审批节点"""
    # 这里会暂停流程，等待人工输入
    approval = human_input.wait_for_input(timeout=300)
    state["approval_log"].append(f"{datetime.now()}: 人工审批结果={approval}")
    state["decision"] = approval
    return state

每个Node职责单一，测试时可以单独mock。更重要的是，流程走到哪个Node、State是什么、耗时多少，都可以打点上报——这是Demo阶段完全不需要考虑、但上线后必须解决的。

---

CSDN资料领取方式

条件分支：业务逻辑的边界

Demo里流程是线性的，生产环境必须处理分支。LangGraph的Edge支持条件路由：

def route_decision(state: AgentState) -> str:
    if state["decision"] == "reject":
        return "deny"
    elif state["decision"] == "pending":
        return "approve_route"
    else:
        return "error"

graph.add_conditional_edges(
    "check_permission",
    route_decision,
    {
        "deny": END,
        "approve_route": "ask_human",
        "error": "handle_error"
    }
)

这里有个实战判断：条件分支不要超过3个。如果路由逻辑复杂到需要写大量if-else，说明Node划分有问题，应该拆成更细的节点。

另一个常见错误：把业务规则硬编码在Edge里。像上面route_decision函数，如果规则会变（比如退款金额阈值调整），应该把规则外置到配置或数据库，Node只负责查配置、做判断。

---

人工审批节点：权限隔离的关键

业务方最在意的点：关键操作必须人工确认。LangGraph支持暂停图执行，等待外部输入：

class HumanApprovalNode:
    def __call__(self, state: AgentState) -> AgentState:
        # 暂停执行，直到收到人工确认
        approval = self.wait_for_human_input(state)
        state["decision"] = approval
        return state

    def wait_for_human_input(self, state: AgentState) -> str:
        # 实际项目中这里可能是：
        # 1. 调用审批API
        # 2. 等待WebSocket消息
        # 3. 轮询数据库状态
        pass

权限隔离的核心：审批节点不应该在Agent进程内完成，而应该调用独立的审批服务。这样即使Agent被攻破，攻击者也无法绕过审批直接执行退款。

我之前犯过的错误：把审批逻辑写在Agent内部，结果测试时发现，只要修改State就能跳过审批。后来改成调用外部审批API，才真正解决了这个问题。

---

工程化落地：日志、重试、可观测

图写完了，离生产还差最后一道坎：工程化。

1. 日志必须打在Node边界

import logging

logger = logging.getLogger(__name__)

def query_order(state: AgentState) -> AgentState:
    logger.info(f"开始查询订单: query={state['user_query']}")
    try:
        order = order_service.search(state["user_query"])
        logger.info(f"查询成功: order_id={order.id}")
        state["order_id"] = order.id
    except TimeoutError:
        logger.warning(f"查询超时: query={state['user_query']}")
        raise  # 让图框架处理重试
    return state

判断标准：如果你的日志打在函数内部而不是边界，排查问题时会非常痛苦。Node是流程的最小单元，日志必须和Node对齐。

2. 重试策略要区分错误类型

from langgraph.types import retry

@retry(max_attempts=3, backoff="exponential")
def query_order(state: AgentState) -> AgentState:
    # ...

不是所有错误都值得重试。网络超时可以重试，参数错误不应该重试。在Node里明确异常类型，比全局加重试更有效。

3. 可观测性：把图执行变成可追踪的事件流

from langchain.callbacks import tracing_v2_enabled

with tracing_v2_enabled() as cb:
    result = graph.invoke(initial_state)
    # cb.events 包含每个Node的执行时间、输入输出、错误信息

生产环境建议接入LangSmith或自建追踪系统。每个Node的执行耗时、State快照、错误堆栈，都应该能被查询和回放。

---