LangChain 中文学习手册

Appearance

案例研究 ​

本页面汇编了团队在生产环境中使用 LangChain 和 LangGraph 的真实案例。通过这些案例，你可以了解常见的 Agent 架构模式、部署策略和最佳实践。

电商客服 Agent ​

背景 ​

某大型电商平台需要构建一个 24/7 在线客服系统，处理订单查询、退款申请、物流追踪和商品推荐等任务。传统基于规则的聊天机器人无法应对复杂的多轮对话和多样化的查询。

架构 ​

用户消息
    │
    ▼
┌──────────────┐     ┌──────────────────┐
│  意图分类节点  │────▶│  订单查询处理节点  │
└──────────────┘     └──────────────────┘
    │                        │
    ├──▶ 退款处理节点         │
    ├──▶ 物流查询节点         │
    ├──▶ 商品推荐节点         │
    └──▶ 人工转接节点         │
                              ▼
                       ┌──────────────────┐
                       │  自然语言生成节点  │
                       └──────────────────┘

实现要点 ​

python
from langgraph.prebuilt import create_agent
from langgraph.graph import StateGraph, END
from typing import Literal

# 意图分类
def intent_classifier(state) -> Literal["order", "refund", "tracking", "recommend", "human"]:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = f"""分类用户意图：{state['messages'][-1].content}
    返回：order | refund | tracking | recommend | human"""
    
    response = llm.invoke(prompt)
    return response.content.strip().lower()

# 构建图
graph = StateGraph(MessagesState)
# ... 添加节点和条件边

取得的效果 ​

  • 首次解决率：从 45% 提升至 82%
  • 平均处理时间：从 8 分钟降至 45 秒
  • 人工转接率：从 55% 降至 18%
  • 客户满意度：提升 23 个百分点

关键经验 ​

  1. 意图分类先行：先分类再处理，避免单一 Agent 负担过重
  2. 人工兜底：设置明确的转人工条件
  3. 情感检测：在关键节点加入情感分析，检测用户情绪变化

智能文档分析平台 ​

背景 ​

一家法律科技公司需要构建一个文档分析平台，能够自动审查合同、提取关键条款、识别潜在风险，并生成合规报告。文档类型包括 PDF、Word 和扫描件。

架构 ​

文档上传
    │
    ▼
┌──────────┐    ┌──────────┐    ┌──────────┐
│ 文档解析   │───▶│ 文本分块   │───▶│ 向量存储   │
└──────────┘    └──────────┘    └──────────┘
                                    │
                                    ▼
                          ┌──────────────────┐
                          │  多 Agent 协作    │
                          │                  │
                          │  ┌────────────┐  │
                          │  │ 条款提取 Agent│  │
                          │  └────────────┘  │
                          │  ┌────────────┐  │
                          │  │ 风险识别 Agent│  │
                          │  └────────────┘  │
                          │  ┌────────────┐  │
                          │  │ 合规检查 Agent│  │
                          │  └────────────┘  │
                          └──────────────────┘
                                    │
                                    ▼
                          ┌──────────────────┐
                          │  报告生成 Agent   │
                          └──────────────────┘

实现要点 ​

python
from langgraph.graph import StateGraph
from langgraph.prebuilt import create_agent

# 创建专业 Agent
clause_extractor = create_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[search_clauses, extract_text],
    system_prompt="你是一个合同条款提取专家..."
)

risk_analyzer = create_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[check_risk_patterns],
    system_prompt="你是一个法律风险分析专家..."
)

compliance_checker = create_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[search_regulations, compare_clauses],
    system_prompt="你是一个合规性检查专家..."
)

# 编排多 Agent 协作
class DocumentAnalysisState(TypedDict):
    documents: List[Document]
    clauses: List[Clause]
    risks: List[Risk]
    compliance_report: str
    final_report: str

workflow = StateGraph(DocumentAnalysisState)
workflow.add_node("extract_clauses", clause_extractor)
workflow.add_node("analyze_risks", risk_analyzer)
workflow.add_node("check_compliance", compliance_checker)
workflow.add_node("generate_report", report_generator)

# 并行执行
workflow.add_edge("extract_clauses", "analyze_risks")
workflow.add_edge("extract_clauses", "check_compliance")
workflow.add_edge("analyze_risks", "generate_report")
workflow.add_edge("check_compliance", "generate_report")
workflow.add_edge("generate_report", END)

取得的效果 ​

  • 文档处理速度：从每份 2 小时降至 5 分钟
  • 条款识别准确率：达到 94%（人工复核后为 99%）
  • 风险发现率：比人工审查多发现 37% 的潜在风险

关键经验 ​

  1. 多 Agent 并行处理：独立任务可以同时执行，大幅提升吞吐量
  2. 分而治之：每个 Agent 专注于单一任务，效果显著优于全能 Agent
  3. 人工在环：关键决策保留人工审核环节

企业数据问答平台 ​

背景 ​

某金融集团需要构建一个内部数据问答平台，员工可以用自然语言查询数据库、Excel 报表和 API 数据。数据源分布在多个部门，格式各异。

架构 ​

用户问题
    │
    ▼
┌──────────────┐
│  问题理解节点   │
└──────────────┘
    │
    ▼
┌──────────────┐
│  数据源路由节点  │──▶ SQL 数据库
└──────────────┘──▶ Excel 报表
         │        ──▶ REST API
         │        ──▶ 实时数据流
         ▼
┌──────────────┐
│  查询执行节点   │
└──────────────┘
    │
    ▼
┌──────────────┐
│  结果融合节点   │
└──────────────┘
    │
    ▼
┌──────────────┐
│  答案生成节点   │
└──────────────┘

实现要点 ​

python
from langgraph.graph import StateGraph
from langgraph.prebuilt import create_agent

# 定义工具
@tool
def query_sales_db(question: str) -> str:
    """查询销售数据库。"""
    sql_agent = create_sql_agent("sales_db")
    return sql_agent.invoke({"question": question})["answer"]

@tool
def query_finance_excel(question: str) -> str:
    """查询财务 Excel 报表。"""
    return pandas_agent.invoke(question)

@tool
def query_crm_api(customer_id: str) -> str:
    """通过 CRM API 查询客户信息。"""
    response = requests.get(f"https://crm.internal/api/{customer_id}")
    return response.json()

# 数据源路由 Agent
router_agent = create_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[query_sales_db, query_finance_excel, query_crm_api],
    system_prompt="""你是企业数据查询助手。根据用户问题：
1. 判断需要查询哪个数据源
2. 如果涉及多个数据源，分别查询后融合结果
3. 给出清晰的答案并注明数据来源"""
)

# 多数据源融合
@tool
def cross_query(sales_data: str, finance_data: str) -> str:
    """融合来自不同数据源的信息。"""
    fusion_prompt = f"""融合以下数据源的信息：
    销售数据：{sales_data}
    财务数据：{finance_data}
    给出综合分析。"""
    return llm.invoke(fusion_prompt)

# 编译图
workflow = StateGraph(QueryState)
workflow.add_node("router", router_agent)
workflow.add_node("fusion", fusion_node)
workflow.set_entry_point("router")
workflow.add_edge("router", "fusion")
workflow.add_edge("fusion", END)

取得的效果 ​

  • 数据查询效率：分析师查询时间从平均 2 小时降至 10 分钟
  • 自助服务率：70% 的数据查询由非技术人员自助完成
  • 数据源统一：整合了 6 个部门共 15 个数据源

关键经验 ​

  1. 数据源抽象：为每个数据源封装独立的工具接口
  2. 权限控制：Agent 只能查询有权限的数据
  3. 查询审计：记录所有 SQL 和 API 调用日志

自动化工作流编排 ​

背景 ​

一家 SaaS 公司需要构建一个智能客服工单处理系统，能够自动分类工单、分配处理人、生成回复草稿，并在必要时升级到高级支持。

架构 ​

工单创建
    │
    ▼
┌──────────────┐
│  工单分类节点   │──▶ 技术支持队列
└──────────────┘──▶ 财务支持队列
         │        ──▶ 产品反馈队列
         ▼
┌──────────────┐
│  自动回复节点   │
└──────────────┘
    │
    ▼
┌──────────────┐
│  升级判断节点   │──▶ 人工处理
└──────────────┘
    │
    ▼
┌──────────────┐
│  工单关闭节点   │
└──────────────┘

实现要点 ​

python
from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver  # 生产级持久化

# 生产级持久化检查点
connection_string = "postgresql://user:pass@host:5432/langgraph"
checkpointer = PostgresSaver.from_conn_string(connection_string)

# 工单处理 Agent
ticket_agent = create_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[
        classify_ticket,
        search_knowledge_base,
        generate_reply,
        assign_to_team,
        escalate_to_human
    ],
    system_prompt="""你是一个智能工单处理系统。处理流程：
1. 分类工单类型和优先级
2. 搜索知识库寻找解决方案
3. 生成回复草稿
4. 判断是否需要升级到人工
5. 记录处理日志""",
    checkpointer=checkpointer        # 持久化状态
)

# 处理逻辑流程图
workflow = StateGraph(TicketState)

workflow.add_node("classify", classify_node)
workflow.add_node("auto_reply", auto_reply_node)
workflow.add_node("escalation_check", escalation_check_node)
workflow.add_node("close", close_node)

workflow.set_entry_point("classify")
workflow.add_edge("classify", "auto_reply")

# 条件：升级或关闭
workflow.add_conditional_edges(
    "auto_reply",
    should_escalate,
    {
        True: "escalation_check",      # 升级判断
        False: "close"                 # 自动关闭
    }
)

workflow.add_edge("escalation_check", "close")
workflow.add_edge("close", END)

# 使用 Postgres 检查点的持久化 Agent
compiled_agent = workflow.compile(checkpointer=checkpointer)

取得的效果 ​

  • 自动处理率：65% 的工单无需人工干预
  • 响应时间：从平均 4 小时降至 5 分钟
  • 准确率：自动分类准确率达 91%

关键经验 ​

  1. 持久化生产状态：使用 PostgresSaver 等生产级检查点，而非 MemorySaver
  2. 条件分支：自动处理 vs 人工升级的判定边界要清晰
  3. 监控与告警：设置 Agent 决策的异常检测

常见架构模式总结 ​

模式适用场景核心组件LangGraph 实现
单 Agent简单问答、单一工具一个 LLM + 工具集create_agent
多 Agent 协作复杂任务、多领域多个专业 Agent + 协调者StateGraph + 条件边
流水线固定流程、批处理顺序执行的节点StateGraph + 串联边
循环/反射需要纠错或多轮推理循环节点 + 条件边条件边的循环回路
并行执行独立子任务扇出 + 扇入扇出边 + 扇入点
人在环 (HITL)审核、批准中断节点 + 人工确认interrupt_before
分层管理大型系统上层 Agent 协调下层嵌套 StateGraph

生产部署最佳实践 ​

检查点持久化 ​

python
# 开发时使用内存
from langgraph.checkpoint.memory import MemorySaver

# 生产环境使用数据库
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.redis import RedisSaver

# Postgres（推荐）
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@host:5432/langgraph"
)
await checkpointer.setup()  # 初始化表

# Redis（低延迟）
checkpointer = RedisSaver.from_conn_string(
    "redis://localhost:6379"
)

监控与可观测性 ​

python
# LangSmith 集成
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-api-key"

# 自定义回调
from langchain_core.callbacks import BaseCallbackHandler

class MetricsHandler(BaseCallbackHandler):
    def on_llm_end(self, response, **kwargs):
        latency = response.response_metadata.get("token_usage", {})
        # 发送到监控系统（Prometheus、Datadog 等）
        metrics.record_llm_latency(latency)

# 使用
llm = ChatOpenAI(callbacks=[MetricsHandler()])

错误处理与重试 ​

python
from langgraph.graph import StateGraph

def robust_node(state):
    try:
        return process(state)
    except RetryableError:
        state["retries"] += 1
        if state["retries"] < 3:
            return state  # 重试
        raise
    except FatalError:
        state["error"] = "不可恢复的错误"
        return state

# 在图中添加超时
workflow.add_node("slow_node", with_timeout(slow_node, timeout=30))

总结 ​

通过这些真实案例，可以看到 LangChain 和 LangGraph 在生产环境中的强大能力：

  1. 灵活编排：StateGraph 提供了无与伦比的控制粒度
  2. 状态管理：内置的检查点系统简化了持久化和恢复
  3. 可扩展性：从简单 Agent 到复杂的多 Agent 系统均可支持
  4. 可观测性：与 LangSmith 等工具集成，方便调试和监控

下一步 ​

本站为非官方中文学习站点，不代表 LangChain 官方。部分内容参考官方文档并重新整理为中文学习笔记。

Copyright © 2026 LangChain 中文学习手册