LangChain 与 LangGraph 生产级实践：智能体架构、状态管理、评估与安全治理

dddddsxh

656人浏览 · 2026-08-08 21:04:23

dddddsxh · 2026-08-08 21:04:23 发布

很多大模型项目的问题，不是模型不会回答，而是系统无法解释、无法恢复、无法评估，也无法在真实业务中承担责任。
真正的智能体系统，核心并不是“让模型多调用几个工具”，而是让模型在明确的状态、流程、权限和反馈机制下工作。

一、LangChain 与 LangGraph 分别解决什么问题？

LangChain 更像是大模型应用的组件生态，主要负责：

模型调用
Prompt 管理
工具封装
RAG 检索
输出解析
多种数据源和模型集成

LangGraph 关注的是更复杂的运行时问题：

多步骤工作流
状态保存
条件分支
循环和重试
人工审批
中断与恢复
多智能体协作

可以简单理解为：

LangChain：提供智能体所需要的零部件

LangGraph：决定这些零部件如何按照流程协作

传统 Chain 通常是线性的：

输入 -> Prompt -> 模型 -> 输出

而生产级 Agent 更接近：

输入
  -> 意图识别
  -> 路由
  -> 检索或工具调用
  -> 生成答案
  -> 风险检查
  -> 人工审批
  -> 输出

这也是 LangGraph 的价值所在。

二、一个生产级 Agent 应该具备什么能力？

一个可以上线的智能体，至少应该满足以下条件：

能够明确知道当前处于哪个步骤
能够保存和恢复执行状态
能够根据不同条件选择不同路径
能够在高风险操作前暂停
能够记录检索来源和执行轨迹
能够对回答质量进行评估
能够限制工具权限和执行范围

因此，Agent 的核心状态不应该只是聊天记录，而应该是一个结构化对象：

{
    "query": "用户问题",
    "route": "knowledge",
    "context": ["检索结果"],
    "answer": "模型草稿",
    "risk": "low",
    "approved": False,
    "citations": ["制度文档"]
}

聊天记录只是状态的一部分。

三、安装依赖

pip install -U langchain langgraph langchain-openai langchain-text-splitters

设置模型密钥：

# Windows PowerShell
$env:OPENAI_API_KEY="your-api-key"

# Linux 或 macOS
export OPENAI_API_KEY="your-api-key"

下面的代码使用 ChatOpenAI，如果使用其他模型，只需要替换模型初始化部分。

四、使用 LangGraph 构建一个生产级问答 Agent

这个示例包含以下节点：

1. 定义状态

from typing import Annotated, Literal
from typing_extensions import TypedDict

from langchain_core.messages import BaseMessage
from langchain_core.documents import Document
from langgraph.graph.message import add_messages


class AgentState(TypedDict, total=False):
    # 对话消息，add_messages 负责自动合并消息
    messages: Annotated[list[BaseMessage], add_messages]

    # 用户当前问题
    query: str

    # 路由结果
    route: Literal["knowledge", "general"]

    # 检索到的文档
    context: list[Document]

    # 生成的答案
    answer: str

    # 引用来源
    citations: list[str]

    # 风险等级
    risk: Literal["low", "high"]

    # 是否通过人工审批
    approved: bool

    # 风险原因
    review_reason: str

这里有一个非常重要的设计原则：

状态字段应该表达业务事实，而不是简单复制模型输出。

例如，approved 是一个业务决策状态，不能只依赖模型生成的自然语言。

2. 初始化模型和向量库

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore


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

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"
)

documents = [
    Document(
        page_content=(
            "员工国内出差住宿标准：一线城市每晚不超过800元，"
            "其他城市每晚不超过500元。超出标准需要提交特殊审批。"
        ),
        metadata={"source": "company_travel_policy.md"}
    ),
    Document(
        page_content=(
            "差旅报销需要提供发票、出差申请单和行程证明。"
            "缺少任一材料时，财务部门可以退回报销申请。"
        ),
        metadata={"source": "company_reimbursement_policy.md"}
    ),
    Document(
        page_content=(
            "生产环境的数据库删除、权限修改和资金相关操作，"
            "必须经过人工审核后才能执行。"
        ),
        metadata={"source": "company_security_policy.md"}
    )
]

vector_store = InMemoryVectorStore(embeddings)
vector_store.add_documents(documents)

retriever = vector_store.as_retriever(
    search_kwargs={"k": 3}
)

真实项目中，文档通常需要先经过切分：

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=80
)

chunks = splitter.split_documents(documents)
vector_store.add_documents(chunks)

文档切分不是越小越好。

切分过小会导致语义不完整，切分过大又会降低检索准确率。应根据文档类型调整：

API 文档：按照接口或章节切分
产品手册：按照功能模块切分
法律制度：按照条款切分
源代码：按照类、函数和模块切分

3. 问题标准化节点

from langchain_core.messages import HumanMessage


def get_latest_question(messages: list[BaseMessage]) -> str:
    for message in reversed(messages):
        if isinstance(message, HumanMessage):
            return str(message.content)

    return ""


def normalize_node(state: AgentState) -> dict:
    query = get_latest_question(state.get("messages", []))

    return {
        "query": query.strip()
    }

问题标准化可以继续扩展为：

去除无意义前缀
补全上下文
识别用户身份
提取租户信息
过滤敏感数据
统一日期和金额格式

例如：

“那报销呢？”

单独看没有明确含义，但结合上一轮对话，可能指“住宿费用如何报销”。

4. 使用结构化输出进行路由

from pydantic import BaseModel, Field
from langchain_core.messages import SystemMessage


class RouteDecision(BaseModel):
    route: Literal["knowledge", "general"] = Field(
        description="knowledge 表示需要查询企业知识库，general 表示普通对话"
    )


router_model = llm.with_structured_output(RouteDecision)


def route_node(state: AgentState) -> dict:
    decision = router_model.invoke([
        SystemMessage(content=(
            "你是一个问题路由器。"
            "涉及公司制度、流程、报销、权限、安全规范的问题，"
            "必须选择 knowledge。普通闲聊和通用知识选择 general。"
        )),
        HumanMessage(content=state["query"])
    ])

    return {
        "route": decision.route
    }

为什么要使用结构化输出？

因为直接让模型返回：

这个问题应该查询知识库。

程序很难稳定解析。

而结构化输出可以将模型结果约束为：

{
  "route": "knowledge"
}

需要注意的是，路由模型的判断不应该直接决定权限。例如：

模型判断：用户可以删除生产数据库

这不能作为真正的授权依据。

路由可以交给模型，权限必须交给确定性的业务规则。

5. 检索节点

def retrieve_node(state: AgentState) -> dict:
    docs = retriever.invoke(state["query"])

    citations = []
    for doc in docs:
        source = str(doc.metadata.get("source", "unknown"))

        if source not in citations:
            citations.append(source)

    return {
        "context": docs,
        "citations": citations
    }

一个常见错误是只把检索结果拼到 Prompt 中，却不记录来源。

这样会导致：

用户无法验证答案
开发者无法追踪错误
评估系统无法判断引用是否正确
出现幻觉时难以定位原因

生产系统应该把以下信息保留下来：

{
    "source": "company_travel_policy.md",
    "chunk_id": "travel-001",
    "score": 0.89,
    "retrieved_at": "2026-08-08T10:00:00"
}

6. 答案生成节点

from langchain_core.messages import AIMessage


def draft_node(state: AgentState) -> dict:
    docs = state.get("context", [])

    if docs:
        context = "\n\n".join(
            f"[文档{i + 1}]\n{doc.page_content}"
            for i, doc in enumerate(docs)
        )
    else:
        context = "当前没有可用的企业内部文档。"

    prompt = f"""
你是一名严谨的企业知识库助手。

请根据以下资料回答问题：

{context}

用户问题：
{state["query"]}

回答要求：
1. 只能使用资料中明确出现的信息。
2. 如果资料不足，明确说明无法确认。
3. 不要编造制度、金额、时间和权限。
4. 涉及操作执行时，只能给出说明，不能假装已经执行。
5. 回答末尾列出使用的文档来源。
"""

    response = llm.invoke([
        SystemMessage(content="你必须忠实依据资料回答问题。"),
        HumanMessage(content=prompt)
    ])

    return {
        "answer": str(response.content)
    }

这里最重要的一句话是：

如果资料不足，明确说明无法确认。

RAG 系统真正的质量，不是“每个问题都回答”，而是：

在知道答案时准确回答，在不知道答案时诚实拒答。

7. 风险检查节点

不要把所有风险判断都交给大模型。

对于删除、转账、改权限、修改生产数据等操作，应当先通过确定性规则拦截：

HIGH_RISK_TERMS = (
    "删除",
    "转账",
    "退款",
    "修改权限",
    "生产数据库",
    "密码",
    "密钥",
    "delete",
    "transfer",
    "production database"
)


def risk_gate_node(state: AgentState) -> dict:
    text = (
        state.get("query", "") + "\n" +
        state.get("answer", "")
    ).lower()

    matched_terms = [
        term for term in HIGH_RISK_TERMS
        if term.lower() in text
    ]

    if matched_terms:
        return {
            "risk": "high",
            "review_reason": (
                "检测到高风险关键词："
                + ", ".join(matched_terms)
            )
        }

    return {
        "risk": "low",
        "review_reason": ""
    }

这段代码并不能替代完整的安全系统，但它体现了一个基本原则：

模型负责理解，规则负责兜底，权限系统负责最终决定。

五、人工介入与可恢复执行

LangGraph 支持在流程中暂停，等待人工处理后继续执行。

from langgraph.types import interrupt


def human_review_node(state: AgentState) -> dict:
    decision = interrupt({
        "type": "approval_required",
        "reason": state.get("review_reason", ""),
        "query": state["query"],
        "answer": state.get("answer", "")
    })

    approved = False

    if isinstance(decision, dict):
        approved = bool(decision.get("approved", False))

    return {
        "approved": approved
    }

当风险较高时，将流程转入人工审批：

def risk_router(state: AgentState) -> str:
    if state.get("risk") == "high":
        return "human_review"

    return "finalize"


def review_router(state: AgentState) -> str:
    if state.get("approved"):
        return "finalize"

    return "blocked"

人工审批不是简单地弹出一个确认框，而是要保留完整上下文：

{
  "type": "approval_required",
  "reason": "检测到修改生产数据库操作",
  "query": "删除生产环境中的订单数据",
  "answer": "该操作需要管理员审核"
}

审批人应该能够看到：

用户是谁
请求来源是什么
模型准备执行什么
使用了哪些工具
影响范围多大
是否存在敏感数据
是否可以回滚

六、组装完整工作流

from langgraph.graph import StateGraph, START, END


def finalize_node(state: AgentState) -> dict:
    return {
        "messages": [
            AIMessage(content=state.get("answer", "没有生成有效答案。"))
        ]
    }


def blocked_node(state: AgentState) -> dict:
    return {
        "answer": (
            "该请求涉及高风险操作，尚未通过人工审核。"
            "系统未执行任何外部操作。"
        )
    }


workflow = StateGraph(AgentState)

workflow.add_node("normalize", normalize_node)
workflow.add_node("route", route_node)
workflow.add_node("retrieve", retrieve_node)
workflow.add_node("draft", draft_node)
workflow.add_node("risk_gate", risk_gate_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("finalize", finalize_node)
workflow.add_node("blocked", blocked_node)

workflow.add_edge(START, "normalize")
workflow.add_edge("normalize", "route")

workflow.add_conditional_edges(
    "route",
    lambda state: state["route"],
    {
        "knowledge": "retrieve",
        "general": "draft"
    }
)

workflow.add_edge("retrieve", "draft")
workflow.add_edge("draft", "risk_gate")

workflow.add_conditional_edges(
    "risk_gate",
    risk_router,
    {
        "human_review": "human_review",
        "finalize": "finalize"
    }
)

workflow.add_conditional_edges(
    "human_review",
    review_router,
    {
        "finalize": "finalize",
        "blocked": "blocked"
    }
)

workflow.add_edge("blocked", END)
workflow.add_edge("finalize", END)

到这里，我们已经把一个线性的问答流程变成了一个带有：

路由
检索
状态
风险检查
人工审批
条件分支
最终输出

的完整图结构。

七、状态持久化与多轮对话

开发阶段可以使用内存检查点：

from langgraph.checkpoint.memory import MemorySaver


memory = MemorySaver()

app = workflow.compile(
    checkpointer=memory
)

调用时必须传入 thread_id：

from langchain_core.messages import HumanMessage


config = {
    "configurable": {
        "thread_id": "user-1001-session-001"
    }
}

result = app.invoke(
    {
        "messages": [
            HumanMessage(
                content="国内出差住宿费用的标准是什么？"
            )
        ]
    },
    config=config
)

print(result["answer"])
print(result.get("citations", []))

thread_id 的含义不是用户 ID，而是一次可恢复的会话执行标识。

同一个 thread_id 可以让系统恢复之前的状态：

result = app.invoke(
    {
        "messages": [
            HumanMessage(content="那报销时还需要什么材料？")
        ]
    },
    config=config
)

生产环境不应使用 MemorySaver 作为唯一存储，因为进程重启后状态会丢失。

生产部署需要使用持久化 Checkpointer，例如数据库或其他可靠存储，并重点考虑：

状态版本
并发更新
租户隔离
数据过期
敏感字段加密
审批记录保留
失败任务重试

八、人工审批后的恢复执行

高风险请求触发中断后，前端可以展示审批信息：

config = {
    "configurable": {
        "thread_id": "risk-session-001"
    }
}

first_result = app.invoke(
    {
        "messages": [
            HumanMessage(
                content="请删除生产数据库中三个月前的订单"
            )
        ]
    },
    config=config
)

print(first_result)

审批通过后，恢复执行：

from langgraph.types import Command


final_result = app.invoke(
    Command(
        resume={
            "approved": True
        }
    ),
    config=config
)

print(final_result["answer"])

审批拒绝：

final_result = app.invoke(
    Command(
        resume={
            "approved": False
        }
    ),
    config=config
)

需要特别强调：

人工审批节点之后，不应该直接执行任意模型生成的代码。

真正执行工具时，仍然需要进行：

用户授权检查
参数校验
资源范围校验
幂等性检查
审计日志记录
超时控制
回滚或补偿

九、工具调用的安全边界

工具不应该无限开放给模型。

错误做法：

tools = [
    database_tool,
    shell_tool,
    file_system_tool,
    payment_tool
]

这种设计相当于让模型拥有一个不受限制的操作系统入口。

更合理的方式是定义白名单：

ALLOWED_TOOLS = {
    "query_order",
    "search_policy",
    "get_user_profile"
}


def authorize_tool(tool_name: str, user_roles: set[str]) -> bool:
    if tool_name not in ALLOWED_TOOLS:
        return False

    if tool_name == "get_user_profile":
        return "hr" in user_roles or "admin" in user_roles

    return True

对于工具参数也要进行校验：

def validate_order_id(order_id: str) -> str:
    if not order_id.isalnum():
        raise ValueError("订单号格式不合法")

    if len(order_id) > 32:
        raise ValueError("订单号长度非法")

    return order_id

模型输出的参数永远是不可信输入。

十、RAG 系统最容易出现的三个问题

1. 检索到了错误文档

原因可能包括：

文档切分不合理
向量模型不适合当前语言
查询问题过短
没有使用元数据过滤
相似度阈值设置不合理

解决思路：

retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={
        "k": 4
    }
)

企业知识库还应该增加租户和权限过滤：

metadata = {
    "tenant_id": "tenant-a",
    "department": "finance",
    "visibility": "internal"
}

用户没有权限看到的文档，不能因为“相似度高”就返回给模型。

2. 检索到了正确文档，但答案仍然错误

可能是模型：

混合了多个文档的结论
忽略了时间范围
误解了否定条件
把建议当成制度
把历史版本当成当前版本

因此 Prompt 中应该明确：

当多个文档存在冲突时，优先使用发布日期最新且状态为有效的文档。
如果无法判断哪个版本有效，请明确说明存在冲突。

3. 答案看起来正确，但没有依据

解决方式是强制引用：

请在答案结尾输出：

参考来源：
- 文档名称
- 条款或章节

如果系统无法提供来源，就应该降低答案可信等级。

十一、如何评估 Agent，而不是凭感觉测试？

很多项目上线前只问几个问题：

“你好”
“公司报销标准是什么？”
“帮我总结这篇文档”

这不能证明系统可靠。

至少应该建立测试集：

test_cases = [
    {
        "id": "travel-001",
        "question": "一线城市出差住宿标准是多少？",
        "expected_sources": ["company_travel_policy.md"],
        "expected_keywords": ["800元"]
    },
    {
        "id": "refund-001",
        "question": "报销需要哪些材料？",
        "expected_sources": ["company_reimbursement_policy.md"],
        "expected_keywords": ["发票"]
    }
]

编写基础评估函数：

def evaluate_case(case: dict) -> dict:
    config = {
        "configurable": {
            "thread_id": f"eval-{case['id']}"
        }
    }

    result = app.invoke(
        {
            "messages": [
                HumanMessage(content=case["question"])
            ]
        },
        config=config
    )

    answer = result.get("answer", "")
    citations = set(result.get("citations", []))

    source_hit = bool(
        citations.intersection(case["expected_sources"])
    )

    keyword_hit = all(
        keyword in answer
        for keyword in case["expected_keywords"]
    )

    return {
        "id": case["id"],
        "source_hit": source_hit,
        "keyword_hit": keyword_hit,
        "answer": answer
    }


results = [
    evaluate_case(case)
    for case in test_cases
]

for item in results:
    print(item)

基础评估指标可以包括：

路由准确率
检索命中率
引用准确率
答案相关性
答案完整性
拒答准确率
工具调用成功率
人工审批触发准确率
平均响应时间
平均 Token 消耗

不要只使用“模型评审模型”。

LLM-as-a-Judge 可以辅助评估，但不应该成为唯一标准。业务规则、关键词、来源匹配和人工抽检仍然不可替代。

十二、可观测性比 Prompt 更重要

生产系统必须知道一次请求经历了什么：

用户问题
  -> 使用了哪个模型
  -> 选择了哪条路由
  -> 检索了哪些文档
  -> 调用了哪些工具
  -> 花费了多少 Token
  -> 哪一步失败
  -> 是否触发人工审核
  -> 最终结果是什么

可以为每次调用增加元数据：

config = {
    "configurable": {
        "thread_id": "user-1001-session-001"
    },
    "tags": [
        "production",
        "knowledge-agent"
    ],
    "metadata": {
        "tenant_id": "tenant-a",
        "user_id": "user-1001",
        "request_id": "req-20260808-001"
    }
}

日志中不应该直接记录：

用户密码
API Key
身份证号
银行卡号
完整客户隐私信息

应在日志层进行脱敏：

import re


def mask_phone(text: str) -> str:
    return re.sub(
        r"(1\d{2})\d{4}(\d{4})",
        r"\1****\2",
        text
    )

日志的目标不是“记录越多越好”，而是：

在不泄露隐私的前提下，让问题可以被复现和定位。

十三、LangChain 与 LangGraph 的工程分层

一个更清晰的项目结构可以是：

agent_project/
├── app.py
├── graph/
│   ├── state.py
│   ├── nodes.py
│   ├── routes.py
│   └── workflow.py
├── retrieval/
│   ├── ingest.py
│   ├── retriever.py
│   └── filters.py
├── tools/
│   ├── order.py
│   ├── user.py
│   └── permissions.py
├── evaluation/
│   ├── dataset.json
│   └── evaluate.py
└── security/
    ├── policy.py
    └── masking.py

推荐的职责划分：

LangChain：
模型、Prompt、Retriever、Tool、Parser

LangGraph：
State、Node、Edge、Checkpoint、Interrupt

业务代码：
权限、审计、数据过滤、异常处理、指标评估

不要把所有逻辑都塞进一个 Agent 节点中。

一个几百行的“超级 Agent”往往比多个职责清晰的小节点更难维护。

十四、从 Demo 到生产需要跨越的几个阶段

阶段一：验证模型能力

目标是确认：

模型是否理解业务问题
RAG 是否能够召回相关内容
工具调用格式是否正确

这个阶段可以使用简单 Chain。

阶段二：固定工作流

当流程开始出现分支、重试和状态时，引入 LangGraph：

路由 -> 检索 -> 生成 -> 校验 -> 输出

阶段三：增加可恢复能力

加入：

Checkpointer
thread_id
失败重试
人工中断
任务恢复

阶段四：增加治理能力

加入：

权限控制
审计日志
数据脱敏
工具白名单
评估数据集
成本和延迟监控

阶段五：持续迭代

每一次线上错误，都应该沉淀为新的测试样例：

线上问题 -> 归因 -> 新增测试用例 -> 修复 -> 回归评估

这才是 Agent 系统不断变可靠的过程。

十五、几个关键结论

LangChain 解决的是“大模型应用如何组装”的问题。

LangGraph 解决的是“复杂智能体如何可靠运行”的问题。

但它们都不能自动解决：

业务权限
数据安全
结果正确性
生产运维
责任边界

真正成熟的智能体系统，不是让模型拥有更多自由，而是让模型在清晰的边界内完成更多工作。

可以用一句话概括：

LangChain 负责能力，LangGraph 负责流程，业务系统负责约束。

如果一个 Agent 只能在理想输入下回答问题，它只是 Demo。

如果它能够在信息不足时拒答、在流程异常时恢复、在高风险操作前暂停、在出现错误后被评估和追踪，它才真正接近生产系统。

写在最后

大模型应用的竞争，正在从“谁能调用模型”转向“谁能把模型变成可靠的软件系统”。

LangChain 让开发者可以快速构建模型、工具和知识库能力；LangGraph 则进一步把这些能力组织成具有状态、分支、记忆和人工控制的执行图。

但真正决定系统能否上线的，不是模型回答得多么精彩，而是：

是否能解释答案从哪里来
是否能知道什么时候不该回答
是否能在失败后继续运行
是否能限制模型的权限
是否能用数据证明系统正在变好

这才是从 Demo 走向生产的真正分水岭。

MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

API网关的作用？MAI Gateway带你快速理解AI网关核心概念

MCP技术社区

企业AI基础设施架构的路径选择：共享底座与联邦互通的长期权衡

本文系统分析了企业AI基础设施的两条演进路线——联邦互通与共享运行时底座。联邦互通通过MCP等协议实现Agent间标准化连接，适合中小规模、快速试错场景；共享底座则通过统一执行层实现全流程可观测、可审计、可管控，适用于大规模、强合规的企业级部署。文章深入探讨了两者的核心分歧、共享底座的六层架构设计、关键设计取舍、成本收益分析，并提供了渐进式落地路径和决策框架，指出两种模式将长期共存、互补演进，企业