LangChain / LangGraph 练习与自测

想想"用户体验"维度：一个是等全部、一个是一个字一个字出。

• invoke：同步等完整结果返回。适合批量任务、结构化输出、后台脚本。
• stream：流式返回 token。适合面向用户的聊天 UI，减少首字等待时间（TTFT）。

真实场景选择：

• 后端 API 批处理 → invoke
• 聊天界面 → stream（配合 SSE/WebSocket）
• 结构化输出（JSON / Pydantic）→ invoke（流式意义不大）

chain 的输出是什么类型？

chain 的输出是 AIMessage，不是 dict，result["q"] 会报 TypeError。

修法 1：加 parser：

from langchain_core.output_parsers import JsonOutputParser
chain = prompt | llm | JsonOutputParser()

修法 2：直接访问：

print(result.content)

LCEL = LangChain Expression Language，用 | 把 Runnable 串成 chain，左边的输出作为右边的输入。

类比 Linux 管道：单向管道，每一段只关心自己的输入输出。LangChain 中 prompt | llm | parser 就是 dict → Prompt → AIMessage → str 的类型流转。

• 自由 JSON：靠 prompt "求"模型返回结构，偶尔多字段、少引号、Markdown 包裹，要自己容错。
• with_structured_output：走模型原生 tool calling 协议，模型端保证结构合法。Pydantic 校验一次即通过。

生产环境几乎必用结构化输出。唯一不用的场景：需要长文本自由创作。

通过 config.configurable.session_id 区分。每个 session_id 对应一份独立的 ChatMessageHistory。

cfg = {"configurable": {"session_id": "u-001"}}
chain.invoke({"input": "..."}, cfg)

相同 session 共享历史；换 session 从零开始。

想想如果一句话刚好被切成两半会发生什么。

• chunk_size：常用 200-800 字符（中文）或 256-1024 token。取决于文档结构：
  • 散文多 / 概念密集 → 偏小（200-400），提高召回精度。
  • 长代码 / 长流程 → 偏大（800-1200），保证完整上下文。
• chunk_overlap：通常 10-20% 的 chunk_size。作用是防止一句完整的话被切到两个 chunk，导致检索时上下文残缺。

经验值：中文知识库起步用 chunk_size=500, overlap=80。

不能。模型只是返回"我想调用这个函数，参数是 X"的意图（tool_call），真正执行是你的代码干的。

完整流程：

1. 你给模型工具清单 → 模型返回 tool_calls
2. 你的代码解析 tool_calls，调真实函数
3. 把函数结果塞回 messages
4. 再调一次模型，让它基于结果回答

把模型当成"开处方的医生"，你是"抓药的药剂师"。

RunnablePassthrough() 把输入原样传递给下游。

这里 invoke 传进来的字符串 "什么是 RAG？" 会：

• 通过 retriever | format_docs 被检索并格式化，填进 context。
• 同时通过 RunnablePassthrough() 原样塞进 question。

等价于手动写：

inputs = {"context": format_docs(retriever.invoke(q)), "question": q}

Passthrough 让 "同一个输入同时分发到多个下游"变得声明式。

通常有两种实现：

• function_calling 模式（OpenAI / 多数现代模型）：schema 在模型端就约束了，基本不会返回非法结构。
• json_mode / json_schema 模式：偶尔会偏差；LangChain 会尝试反序列化，失败则抛 OutputParserException。

实战应对：

• 加 method="function_calling" 显式走工具调用（如果模型支持）。
• 业务层包一层重试：try/except OutputParserException 后 retry 一次，仍失败才降级。
• 打 metric：结构化失败率超过 1% 就告警。

想想 100 轮对话之后 prompt 长什么样。

• Token 爆炸：每轮都要把全部历史塞进 prompt，100 轮就上万 token，费用和延迟双爆。
• Lost-in-middle：模型对长上下文中段的注意力会显著下降，早期信息被稀释。
• 没有语义检索：历史里有一条关键信息，第 50 轮问到时如果没被截到就完全丢了。

长期记忆正确做法：

• 用 trim_messages 滚动截断（保留最近 N 条 + system）。
• 把关键事实抽成结构化记忆，存进 vector store / kv store，需要时再检索回来。
• LangGraph 用 Store 做跨 thread 的长期记忆。

Bug：messages 字段没有 reducer，LangGraph 的默认合并是覆盖，所以后一个节点把前一个的值覆盖了。

修法：用 add_messages reducer（来自 langgraph.graph.message）：

from langgraph.graph.message import add_messages
from typing import Annotated
 
class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

这样两个节点的返回会被追加合并，最终 ["hi", "bye"]。

想想条件边 + 计数器字段。

class State(TypedDict):
    answer: str
    quality: str
    retry_count: int
 
def quality_check(state) -> str:
    if state["quality"] == "good" or state["retry_count"] >= 3:
        return "end"
    return "retry"
 
def answer_node(state):
    ans = llm.invoke(...)
    return {"answer": ans, "retry_count": state["retry_count"] + 1}
 
def check_node(state):
    q = judge_llm.invoke(state["answer"])  # 返回 good/bad
    return {"quality": q}
 
g.add_node("answer", answer_node)
g.add_node("check", check_node)
g.add_edge(START, "answer")
g.add_edge("answer", "check")
g.add_conditional_edges("check", quality_check, {
    "retry": "answer",   # ← 回跳
    "end": END,
})

关键点：

• 条件边的目标字典里可以指回之前的节点（形成环）。
• 必须有终止条件（retry_count 上限），不然死循环。

选型决策：

• 上线前 dev → SqliteSaver
• 正式生产 → PostgresSaver + 连接池
• 绝不把 InMemorySaver 部到生产

机制：

1. 节点调用 interrupt(payload) 时，LangGraph 把当前 State 写进 checkpointer，把 payload 存进 PendingInterrupt，然后抛出 GraphInterrupt。
2. 外层 app.invoke() 捕获到，不继续执行后续节点，返回当前 State。
3. 外部系统拿到 payload，展示审批 UI。
4. 审批完成后，用 Command(resume=decision) 再次 invoke，同 thread_id。
5. LangGraph 从 checkpointer 读出 State，重新执行那个节点（这次 interrupt() 直接返回 decision）。

为什么必须 checkpointer：State 不持久化就没法"等"——一旦函数返回，State 就丢了。interrupt 本质是"State 持久化 + 函数暂停恢复"。

生产注意点：

• resume 可能在分钟/小时/天级延迟后发生，所以用 PostgresSaver。
• 节点要幂等：因为 resume 后该节点会重跑一遍。副作用代码（比如发邮件）要放在 interrupt 之后。

LangGraph 支持并行节点（同一 superstep 内）。并行节点的写入合并由 reducer 决定：

log: Annotated[list[str], operator.add]

• 并行 A 和 B 同时跑完，框架把它们返回的两个 list 按 superstep 顺序合并（非"严格执行时间"顺序）。
• 同一 superstep 内节点间不保证顺序——所以不能用 log 顺序表达因果。
• 跨 superstep 的顺序是严格的：superstep N 的所有节点都合并完，再进入 superstep N+1。

实战：

• 需要严格顺序 → 串行连边，不要并行。
• 日志只作观察用 → 并行可以。
• 关键因果信息要带显式字段（如 timestamp）。

三层实现：

1. 流式层：用 app.astream(...) 或 app.stream(...) 逐 step 产出，前端能收到 token/step 级事件。
2. 取消信号：前端按 Stop → 后端关闭 SSE/WebSocket → 后端捕获到 client disconnect → 取消当前任务（Python 的 asyncio.CancelledError）。
3. 状态清理：
   • 正在运行的 tool_call 要能响应取消（工具内部要用 asyncio.sleep 而非 blocking）。
   • 已经持久化的 checkpoint 保留，用户再点"继续"时能从断点恢复。
   • 或者标记为 cancelled 状态，下次对话从新开始。

LangGraph 0.2+ 提供 app.astream_events() 可以 cancel task。配合 thread_id 和 checkpointer，甚至可以"时间旅行"回到 cancel 之前的某个 state 重跑。

State：

class State(TypedDict):
    urls: list[str]
    fetched: Annotated[list[Document], operator.add]
    labeled: Annotated[list[dict], operator.add]
    failed_urls: Annotated[list[str], operator.add]
    run_id: str
    stats: dict

节点：

START → plan (生成 url 列表)
     → fetch (并行抓，失败写 failed_urls)
     → clean (去重/去噪)
     → label_llm (调 LLM 打标，结构化输出)
     → validate (校验字段完整)
     → write_db (落库)
     → notify (钉钉/Slack 通报)
     → END

Checkpointer：PostgresSaver，按 run_id 作 thread_id。

失败恢复：

• 每个节点幂等（用 url hash 做 dedup key，INSERT ... ON CONFLICT）。
• 节点失败不吞异常，写进 failed_urls 或抛出到全局捕获。
• 重跑：从 checkpoint 恢复同 run_id，跳过已完成的 url（节点内做幂等判断）。
• 超时：给每个 LLM 调用加 timeout=30，fetch 加 timeout=10。

观测：

• 每个节点打 trace（OpenTelemetry）：开始时间、耗时、输入 hash、输出大小、status。
• 关键 metric：抓取成功率、打标耗时 p95、每日 token 费用、失败 url 数。
• 报警：失败 url > 10% 或总耗时超基线 2 倍。

把现有脚本改成 Agent 的真正好处：

• 任意节点失败可断点续跑（比每次全量重跑省 10x 成本）。
• 人审节点可加（如关键 url 打标 → interrupt → 值班同学确认）。
• State trace 在 LangSmith 可视化，排障从小时级降到分钟级。

排查（必须带 trace）：

1. LangSmith 或自建 trace 里过滤 tool_call_count > 5 的 thread。
2. 看循环调用的工具是哪个、参数长什么样、LLM 的 reasoning 文本是什么。
3. 统计哪些用户问法最容易触发。

常见根因：

• 工具描述不清：LLM 不知道"为什么上一次调用的结果已经够了"。
• 工具返回值不够明确：返回 "未找到" 太含糊，LLM 不死心再换参数试。
• Schema 字段冗余：LLM 在不同字段间反复换，以为"这次会不一样"。
• Prompt 没停止信号：没告诉 LLM "如果工具返回 X 就停止"。

修复五步：

1. 给图加硬上限：全局计数器字段 tool_call_count，超过 5 次强制路由到 end_with_apology 节点。
2. 工具返回值格式标准化：{"status": "ok|not_found|error", "data": ..., "hint": "停止查询"}。
3. 优化工具 description，写清"什么情况不要再调"。
4. 加 LLM-as-Judge 检测循环："看看这个 thread 是不是在循环？"，如是则中断。
5. 在 checkpoint 里保留失败案例，周度对抗集评测确保回归不变坏。

add_messages 不只是简单的 list 追加，它做了三件事：

1. 类型容忍：接受 BaseMessage 子类、dict 形式、甚至字符串，自动规整为 BaseMessage。
2. 按 ID 去重/替换：如果新消息和历史中某条消息 id 相同，则替换而非追加。这是实现"时间旅行"和"消息编辑"的关键。
3. 顺序保持：追加到末尾，保持时间顺序。

对比 operator.add：

• 单纯追加。
• 不去重，同一条消息被多次返回会出现多份。
• 不规整类型（如果节点返回 dict 而不是 BaseMessage 会后续崩）。

进阶用法：通过让节点返回带 id 的 AIMessage，可以"修改历史"——这是 LangGraph Studio 编辑功能的底层机制。

基础设施（3 件）：

1. PostgresSaver + 连接池：不能用 Sqlite/InMemory；connection 至少 20 个起。
2. 模型密钥托管：用 KMS 或 Vault，不硬编码；启动时拉取。
3. 向量库集群：Chroma 换成 Qdrant/Weaviate/Milvus，带备份。

可靠性（3 件）： 4. 超时与重试：每个 LLM 调用 timeout + exponential backoff；工具调用单独限额。 5. 熔断降级：连续失败 N 次自动切备用模型或返回"系统繁忙"模板。 6. 幂等与去重：外部工具（下单、退款）必须用 idempotency_key；已执行的 action 重放不重复。

安全合规（2 件）： 7. Guard 输入输出：Prompt injection 过滤；PII 脱敏后再进向量库（参考 ai-application-security topic）。 8. 审计日志 WORM：谁在 thread X 对话了什么、调了什么工具、审批决策，保留 ≥ 1 年。

可观测（2 件）： 9. Trace + Metrics：每 invoke 打 trace（模型/latency/tokens/cost）；dashboard 追 p50/p95、token 花销、工具成功率。 10. 评测与回归：黄金集 50-100 条，每次 prompt/model 变更自动跑评测，分数低于阈值拦发布（参考 llmops-observability-evaluation topic）。

加分项：

• interrupt 的审批通知链（钉钉/邮件）+ 超时未审核兜底策略。
• 流式 SSE 接口 + 前端 cancel 路径。
• 灰度 1% → 10% → 100% 的放量计划，配回滚开关。