《大家都在聊LangGraph，企业真正需要的却不是更多 Demo》看起来是个大话题，但真落到项目里，常常就是几个具体选择。下面我尽量按实际开发时会遇到的问题来讲。

摘要

摘要：上周复盘一个 Agent 项目，业务方一句话"跨文档关联"就把整个团队打回原形。我们之前用 LangChain 写的脚本式 Agent 在 Demo 里跑得挺顺，但真正要接入生产环境时，权限校验、日志追踪、人工审批这些环节全乱了。这篇文章复盘我们用 LangGraph 重构工作流的过程，聊聊从 Demo 到生产系统真正需要补的功课。

---

目录

• 为什么脚本式 Agent 在生产环境会崩
• State 与 Node：把状态显式化
• Edge 与条件分支：让流程可控
• 人工审批节点：权限校验的正确姿势
• 工程化落地：日志、追踪和可观测
• 总结

---

为什么脚本式 Agent 在生产环境会崩

我先说一个真实场景。

去年我们团队接了一个内部知识库问答的 Agent 需求，用的是 LangChain 的 ChatChain 写法，大概 200 行代码就搭出了一个能跑的 Demo。业务方试用后说"挺好用"，我们就以为能上线了。

结果上线前一周，安全团队来评审，问了三个问题：

1. Agent 调用内部 API 时，有没有做权限校验？
2. 每次请求的日志能不能追踪到具体是哪个用户、什么时间、问了什么？
3. 涉及敏感数据的查询，需不需要人工审批？

我们当时愣住，因为 Demo 里根本没考虑这些东西。

回看代码，问题出在架构上。我们用的是典型的脚本式写法：

# 典型的脚本式 Agent（有问题）
def run_agent(user_question):
    # 1. 直接调用 LLM
    response = llm.invoke(user_question)

    # 2. 根据回复决定下一步
    if "查询" in response:
        result = call_internal_api(user_question)
    elif "审批" in response:
        result = send_approval_request(user_question)
    else:
        result = response

    return result

这段代码在 Demo 里能跑，但有几个致命问题：

状态不可追踪。每次调用都是独立的，中间过程没有记录，出问题的时候根本不知道 Agent 走到哪一步了。

流程不可控。条件分支硬编码在函数里，想加一个新类型的问题处理，就得改代码、重新部署。

权限和日志是事后补的。等安全团队来问，才发现根本没设计。

这就是为什么很多团队说"Agent 能跑 Demo，但上不了线"。不是模型不行，是架构不行。

---

State 与 Node：把状态显式化

LangGraph 的核心思想是把 Agent 的工作流显式化成一个图。每个节点（Node）是一个函数，图的状态（State）是节点之间传递的数据。

我们重构后的代码长这样：

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

# 定义 State，所有状态都显式声明
class AgentState(TypedDict):
    user_question: str
    user_id: str
    intent: str  # 识别出的意图
    response: str  # LLM 回复
    api_result: dict  # API 调用结果
    approval_status: str  # 审批状态
    final_answer: str  # 最终答案
    logs: list  # 日志记录

# 每个节点只负责一件事
def intent_recognition_node(state: AgentState) -> AgentState:
    """识别用户意图"""
    intent = classify_intent(state["user_question"])
    state["intent"] = intent
    state["logs"].append({
        "step": "intent_recognition",
        "intent": intent,
        "timestamp": datetime.now().isoformat()
    })
    return state

def llm_response_node(state: AgentState) -> AgentState:
    """调用 LLM 生成回复"""
    prompt = build_prompt(state["user_question"], state["intent"])
    response = llm.invoke(prompt)
    state["response"] = response
    state["logs"].append({
        "step": "llm_response",
        "response_length": len(response),
        "timestamp": datetime.now().isoformat()
    })
    return state

def api_call_node(state: AgentState) -> AgentState:
    """调用内部 API"""
    result = call_internal_api(state["user_question"], state["user_id"])
    state["api_result"] = result
    state["logs"].append({
        "step": "api_call",
        "success": result.get("success"),
        "timestamp": datetime.now().isoformat()
    })
    return state

关键点：

State 是显式的。所有中间状态都在 TypedDict 里声明，谁都能看懂 Agent 当前走到哪一步、数据是什么。

节点是纯函数。输入 State，输出 State，没有副作用。这样的好处是节点可以独立测试，出了问题也容易定位。

日志内建在 State 里。每次节点执行都会往 logs 列表里追加一条记录，上线后直接查 State 就能拿到完整执行轨迹。

---

Edge 与条件分支：让流程可控

脚本式 Agent 的流程是硬编码的，LangGraph 用 Edge 把流程变成可配置的图。

# 构建图
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("intent_recognition", intent_recognition_node)
graph.add_node("llm_response", llm_response_node)
graph.add_node("api_call", api_call_node)
graph.add_node("approval", approval_node)
graph.add_node("format_answer", format_answer_node)

# 设置入口
graph.set_entry_point("intent_recognition")

# 条件分支：根据意图决定下一步
def route_by_intent(state: AgentState) -> str:
    if state["intent"] == "query":
        return "api_call"
    elif state["intent"] == "approval_required":
        return "approval"
    else:
        return "format_answer"

graph.add_conditional_edges(
    "llm_response",
    route_by_intent,
    {
        "api_call": "api_call",
        "approval": "approval",
        "format_answer": "format_answer"
    }
)

# 普通边
graph.add_edge("intent_recognition", "llm_response")
graph.add_edge("api_call", "format_answer")
graph.add_edge("approval", "format_answer")
graph.add_edge("format_answer", END)

# 编译图
app = graph.compile()

条件分支的好处是：加新流程不需要改代码。比如业务方突然说"查询类请求需要二次确认"，我们只需要：

1. 加一个 secondary_confirm 节点
2. 修改条件路由函数
3. 加一条条件边

State 和 Node 的设计让流程变更变得可预测，而不是到处打补丁。

---

人工审批节点：权限校验的正确姿势

回到最初的问题：权限校验怎么做？

我们之前的做法是在 API 调用前加一个 if 判断，但这有几个问题：判断逻辑散落在代码里，不好统一管理；审批流程不可追踪；无法处理需要人工介入的场景。

用 LangGraph 的方式，我们把审批做成一个独立的节点：

def approval_node(state: AgentState) -> AgentState:
    """人工审批节点"""
    # 1. 生成审批请求
    approval_request = {
        "user_id": state["user_id"],
        "question": state["user_question"],
        "intent": state["intent"],
        "risk_level": assess_risk(state)
    }

    # 2. 发送审批请求（比如发到企业微信/钉钉）
    send_approval_request(approval_request)

    # 3. 等待审批结果（这里用轮询，生产环境建议用消息队列）
    approval_result = wait_for_approval(approval_request["id"])

    # 4. 根据审批结果更新 State
    if approval_result["status"] == "approved":
        state["approval_status"] = "approved"
        state["logs"].append({
            "step": "approval",
            "status": "approved",
            "approver": approval_result["approver"],
            "timestamp": datetime.now().isoformat()
        })
    else:
        state["approval_status"] = "rejected"
        state["final_answer"] = "您的请求已被拒绝，原因：" + approval_result["reason"]
        state["logs"].append({
            "step": "approval",
            "status": "rejected",
            "reason": approval_result["reason"],
            "timestamp": datetime.now().isoformat()
        })

    return state

# 权限校验也可以做成节点
def permission_check_node(state: AgentState) -> AgentState:
    """权限校验节点"""
    user_permissions = get_user_permissions(state["user_id"])
    required_permission = get_required_permission(state["intent"])

    if required_permission not in user_permissions:
        state["final_answer"] = "您没有权限执行此操作"
        state["logs"].append({
            "step": "permission_check",
            "passed": False,
            "user_id": state["user_id"],
            "timestamp": datetime.now().isoformat()
        })
        return state

    state["logs"].append({
        "step": "permission_check",
        "passed": True,
        "user_id": state["user_id"],
        "timestamp": datetime.now().isoformat()
    })
    return state

关键设计：

审批和权限校验都是节点。这意味着它们可以被独立测试、独立监控、独立调整。

审批结果写入 State。后续节点可以根据审批状态决定怎么做，整个流程是透明的。

日志内建。谁在什么时候审批了什么，全部记录在 State 里，上线后直接查。

---

工程化落地：日志、追踪和可观测

有了图工作流，工程化问题就清晰了。我们当时踩的坑主要是三个：

1. 日志分散，排查困难

脚本式 Agent 的日志散落在各个函数里，出了问题要翻好几层代码。用 LangGraph 后，所有日志都在 State 的 logs 字段里，一条查询就能拿到完整执行轨迹：

# 执行 Agent
result = app.invoke({
    "user_question": "查询上月销售数据",
    "user_id": "user_123",
    "intent": "",
    "response": "",
    "api_result": {},
    "approval_status": "",
    "final_answer": "",
    "logs": []
})

# 直接查日志
for log in result["logs"]:
    print(f"[{log['timestamp']}] {log['step']}: {log}")

2. 追踪断点

脚本式 Agent 跑挂了，不知道是 LLM 的问题还是 API 的问题。用 LangGraph 后，每个节点的状态都在 State 里，可以直接看到 Agent 走到哪一步、哪一步失败了：

# 检查 State 中的中间状态
print(f"意图识别: {result['intent']}")
print(f"审批状态: {result['approval_status']}")
print(f"API结果: {result['api_result']}")

3. 可观测性

上线后我们需要监控 Agent 的健康状况。用 LangGraph 后，可以在每个节点前后加 hook，统一收集指标：

from langgraph.prebuilt import create_react_agent
import time

# 在节点执行前后记录耗时
def timed_node(node_func):
    def wrapper(state):
        start = time.time()
        result = node_func(state)
        elapsed = time.time() - start
        result["logs"].append({
            "step": f"{node_func.__name__}_duration",
            "duration_seconds": elapsed,
            "timestamp": datetime.now().isoformat()
        })
        return result
    return wrapper

# 给节点加计时
graph.add_node("llm_response", timed_node(llm_response_node))

---

总结

从 Demo 到生产，Agent 项目真正要补的不是模型调用能力，而是工程化能力。具体来说：

状态要显式化。用 State 而不是局部变量，所有中间状态都能追踪。

流程要图化。用 Edge 而不是 if-else，流程变更可配置、可测试。

权限和审批要节点化。做成独立节点，统一管理和监控。

日志要内建。每个节点执行都记录日志，出问题能快速定位。

我们团队用 LangGraph 重构后，上线前安全评审一次性通过。不是因为模型变强了，是因为架构变清晰了。

如果你也在做 Agent 项目，建议尽早用图工作流的思路重构。Demo 能跑只是第一步，能上线、可维护、可观测才是真本事。

资料展示

下面是我整理的AI大模型学习资料和工具包预览，适合收藏后按主题逐步学习。

如果你想看完整资料目录，可以在评论区留言「资料」；也欢迎告诉我你更关注AI大模型里的哪类内容。