第 9 篇立完回归纪律，系列攒下的能力清单已经齐整：模型入口、提示模板、结构化输出、工具调用、LCEL 链、agent、双记忆、RAG、评测——九块积木各自单独验过收，却还没在同一个程序里同框工作过。本篇是《LangChain 从入门到实战》第 10 篇，也是收官篇：把它们全部装进一个真实项目——知识库客服机器人，从需求一路走到评测验收。单看任何一行代码都是旧知识，合在一起是一台能交付的机器。
项目设定：云看板是一款虚构的团队协作看板工具，代指任何有产品文档的软件，官方文档以 markdown 形式放在 docs 目录。客服机器人要答准产品问题、答不了转工单、同会话连续追问不失忆、跨会话记住用户偏好。本文按"需求与架构 → 模块走读 → 运行实录 → 评测验收 → 对照与扩展"组织全文；依赖沿用第 1、8、9 篇的口径：主包加 langchain-openai 与 langchain-text-splitters，评测另装 langsmith。
1、需求与总体架构
把"产品文档客服"翻译成五条可验收的需求，每条正好对应系列的一块积木，对齐如下：
| 需求 | 能力 | 回扣篇目 |
|---|---|---|
| 产品问题答得准 | RAG 检索 | 第 8 篇 |
| 答不了转人工 | 工单工具 | 第 4 篇 |
| 同一会话连续追问 | 会话记忆 | 第 6 篇 |
| 跨会话记住偏好 | 长期记忆 | 第 7 篇 |
| 改动有分数背书 | 评测验收 | 第 9 篇 |
五行缺一都会露馅：缺检索会编造、缺工单用户卡死、缺会话记忆用户每轮自报家门、缺长期记忆偏好记不住、缺评测调优回到玄学。五条需求同样决定了架构的三个层次：前两条是工具层，中间两条是记忆层，最后一条是流程层。整体架构如下：
图里三条线索值得点透。知识库建设在左侧离线跑：文档切块入库一次，检索工具与评测共用同一个向量库，文档更新只需重跑这条线、问答服务不动——第 8 篇"两段分离"的工程意义落到项目里。中间是装配：四个工具交给 create_agent，checkpointer 与 store 双记忆挂在两侧，分别管会话历史与用户偏好。右下是闭环：评测直接打在检索层，分数不达标就回去调参——第 9 篇的回归纪律第一次以代码形式进入项目。
2、知识库切块入库
索引段照搬第 8 篇五步的前四步：加载用原生构造 Document（多文件遍历比加载器直白），切分用 RecursiveCharacterTextSplitter，嵌入入库用 InMemoryVectorStore，完整代码如下：
from pathlib import Path
from langchain_core.documents import Document
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_core.vectorstores import InMemoryVectorStore
docs = [Document(page_content=md.read_text(encoding="utf-8"),
metadata={"source": md.name})
for md in Path("docs").glob("*.md")]
splitter = RecursiveCharacterTextSplitter(chunk_size=500,
chunk_overlap=80)
splits = splitter.split_documents(docs)
vector_store = InMemoryVectorStore(
OpenAIEmbeddings(model="text-embedding-3-large"))
vector_store.add_documents(splits) # 嵌入入库一行完成
retriever = vector_store.as_retriever(search_kwargs={"k": 4})
参数手感沿用第 8 篇的结论：产品单篇文档不长，chunk_size 取 500、重叠 80；k 先取 4，第 6 节的评测会让这个数字接受审判。有一处伏笔：metadata 里的 source 一物两用——既在检索结果里给答案标注出处，又是命中率评估器的断言依据。retriever 在这一步就建好，工具层与评测层共用同一份索引，避免验收与上线两套索引；切块参数的取舍见官方 RAG 教程。
3、定义四个工具
工具是 agent 的手脚，描述是模型路由的唯一依据（第 4 篇），四个工具的 docstring 都按"做什么、什么时候用"写足信息量。
3.1、检索与工单工具
检索工具把第 8 篇的检索器与第 5 篇的管道收进一个函数；工单工具用第 3 篇的 Pydantic schema 锁死参数。检索与工单两个工具的定义如下：
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.runnables import RunnableLambda
from langchain_core.tools import tool
def format_docs(hits): # 检索块拍平成带来源的文本
return "\n\n".join(f"[{h.metadata['source']}]\n{h.page_content}"
for h in hits)
retrieve = retriever | RunnableLambda(format_docs) # 第 5 篇的管道
@tool
def search_docs(question: str) -> str:
"""按问题语义检索云看板产品文档，返回带来源的片段。
回答功能、价格、操作步骤前必须先调用本工具。"""
return retrieve.invoke(question)
tickets = [] # 演示用，生产接工单系统
class TicketInput(BaseModel):
summary: str = Field(description="一句话概括用户遇到的问题")
priority: Literal["低", "中", "高"] = Field(
default="中", description="功能不可用或用户明确不满取高")
@tool(args_schema=TicketInput)
def create_ticket(summary: str, priority: str = "中") -> str:
"""检索结果无法解决用户问题时，创建人工工单转人工跟进。"""
return f"已建工单 TK-{1000 + len(tickets)}（{priority}），1 个工作日内跟进。"
四点解读。retrieve 一行是第 5 篇 LCEL 的直接复用：检索器天生是 Runnable，与 RunnableLambda 串成管道，工具函数里只剩一次 invoke。search_docs 的描述写明"必须先调用"，这是给路由上的第一道约束——没有它，模型会拿训练记忆硬答产品问题。TicketInput 的 priority 用 Literal 枚举加描述，"功能不可用或用户明确不满取高"这句话就是模型判优先级的全部依据，第 3 篇"描述写给模型看、约束留给校验器"在工具上的落点。tickets 是演示替身，换成工单系统 API 时签名一行不用改。
安全提示：建工单是写操作。描述里"解决不了再建"与系统提示的双保险之外，生产上还应加调用限流与审计日志——工具的执行权在你手里，第 4 篇说的安全边界落到真实项目就是这一行防御。
3.2、偏好记忆工具
长期记忆按第 7 篇官方推荐的模式做：读写都是工具，记什么、何时记由 agent 决定，runtime 注入 store 与用户身份。两个记忆工具的定义如下：
from dataclasses import dataclass
from langchain.tools import ToolRuntime # tool 沿用 3.1 的导入
@dataclass
class Context: # 请求级上下文：谁在说话
user_id: str
@tool
def save_preference(preference: str,
runtime: ToolRuntime[Context]) -> str:
"""把用户明确表达的偏好写入长期记忆，如回复要简短。"""
runtime.store.put(("preferences", runtime.context.user_id),
preference, {"text": preference})
return "已记住。"
@tool
def recall_preferences(runtime: ToolRuntime[Context]) -> str:
"""读出该用户已记住的全部偏好，回答前先调用。"""
items = runtime.store.search(
("preferences", runtime.context.user_id))
return "；".join(it.value["text"] for it in items) or "暂无记录"
三点解读。runtime 参数由框架注入，不进模型可见的签名——模型看到的 save_preference 只有一个 preference 参数。偏好写入的命名空间是 ("preferences", user_id)，与线程号无关，任何会话召回的都是同一份，第 7 篇"checkpointer 管一个线程、Store 管一个人"的分工在此落地。recall_preferences 的描述同样写了调用时机，与 3.1 的路由约束同一手法；这套模式的官方口径见长期记忆文档。
4、装配双记忆 agent
4.1、create_agent 装配
四件东西拼装：模型、四个工具、系统提示、双记忆。装配代码如下：
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
model = init_chat_model("openai:gpt-6.1-terra", temperature=0)
agent = create_agent(
model,
[search_docs, create_ticket,
save_preference, recall_preferences],
system_prompt="你是云看板的客服专员：产品问题先检索文档，"
"资料不足就明说，不要编造；检索解决不了再建工单；"
"用户表达偏好先记住再回答。",
checkpointer=InMemorySaver(), # 短期记忆：会话内历史
store=InMemoryStore(), # 长期记忆：跨会话偏好
context_schema=Context, # 请求上下文：用户身份
)
解读：create_agent 一行收编执行循环（第 6 篇），checkpointer 管会话内历史、store 管跨会话偏好（第 7 篇），context_schema 让用户身份随每次调用传入而不占模型上下文。系统提示按第 2 篇的思路收成常量：人设、护栏、流程各一句，改口径只动这一处。参数语义以官方 Agents 文档为准。
双记忆此处都是内存版：
InMemorySaver与InMemoryStore重启即失。生产按第 7 篇换SqliteSaver、PostgresSaver与PostgresStore（见持久化文档），装配代码一个参数都不用改。
4.2、主服务循环
主循环是十几行 CLI：每会话一个 thread_id，每个用户一个 context。代码如下：
def chat(session: str, user: str):
cfg = {"configurable": {"thread_id": session}}
ctx = Context(user_id=user) # 身份不进模型上下文
print(f"进入会话 {session}（用户 {user}），输入 q 退出")
while True:
q = input("你：")
if q.strip().lower() == "q":
break
r = agent.invoke(
{"messages": [{"role": "user", "content": q}]},
config=cfg, context=ctx)
print("客服：", r["messages"][-1].content)
chat("s-001", "user-001")
三点解读。cfg 与 ctx 在会话开始时构造一次、循环里原样复用——thread_id 隔离会话，user_id 隔离偏好，两个键互不纠缠。invoke 的三参形态是第 7 篇的写法：消息、config、context 各归各位。取最终回答依旧是 r["messages"][-1].content（第 6 篇的读法）；要亲眼看每轮调了哪些工具，把 invoke 换成 stream(stream_mode="updates") 打一遍轨迹即可。
5、三段对话运行实录
python cs_bot.py 启动，先在会话 s-001 连问三句，再换会话 s-002 验证长期记忆，一次典型运行如下：
进入会话 s-001（用户 user-001），输入 q 退出
你：云看板支持离线使用吗？
客服：支持。移动端在设置里开启"离线模式"，可查看并编辑最近
打开的看板，联网后自动同步；桌面端暂不支持离线编辑。
你：我配了 Webhook，收不到推送，帮我处理一下。
客服：文档里的常见原因都排查过：回调地址未验证、签名不匹配、
防火墙拦截，均不符合你的情况。已创建工单 TK-1001（优先级高），
1 个工作日内有同事联系你核对配置。
你：以后回复都简短一点，帮我记住。
客服：已记住：回复保持简短。
进入会话 s-002（用户 user-001），输入 q 退出
你：免费版最多几个人用？
客服：免费版上限 5 人，更多请升级专业版。
三段各验一层。第一段普通问答：search_docs 被调用，答案细节全部来自文档块，检索结果自带的 [mobile.md] 出处被模型转述成自然语言，可溯源是客服场景的硬指标。第二段建工单是两步工具链：模型先调 search_docs 排查，判断检索解决不了，再调 create_ticket，priority 按 schema 描述取了"高"——用户明确说功能不可用，第 3 篇的字段描述与第 4 篇的描述即路由在此同时兑现。第三段验双记忆：第三问触发 save_preference 当场写入；换到 s-002 后 thread_id 变了、短期记忆清零，但回答只有一句——recall_preferences 从 Store 召回了偏好，长期记忆跨会话生效。
6、黄金集评测验收
第 9 篇的结论直接搬来验收：答案质量的瓶颈几乎总在检索不在生成，先给检索层立命中率护栏。黄金集共 10 问：8 条文档问按"每篇文档至少一问"从真实客服日志分层抽取，2 条越界问考"建工单或明说没有"、用手写断言循环单独验收。走命中率的部分只有 8 条，代码如下：
from langsmith import Client
from langsmith.evaluation import evaluate
golden = [ # 8 条文档问，2 条越界问见正文
{"q": "免费版最多几个成员？", "src": "billing.md"},
{"q": "专业版怎么开票？", "src": "billing.md"},
{"q": "移动端怎么开离线模式？", "src": "mobile.md"},
{"q": "桌面端能离线编辑吗？", "src": "mobile.md"},
{"q": "Webhook 怎么配置？", "src": "webhook.md"},
{"q": "推送收不到怎么办？", "src": "webhook.md"},
{"q": "怎么导出看板数据？", "src": "export.md"},
{"q": "看板能分享给非成员吗？", "src": "faq.md"},
]
client = Client()
ds = client.create_dataset(dataset_name="客服黄金集")
client.create_examples(inputs=[{"question": c["q"]} for c in golden],
outputs=[{"source": c["src"]} for c in golden],
dataset_id=ds.id)
def retrieve_target(inputs: dict) -> dict: # 只测检索不测对话
hits = retriever.invoke(inputs["question"])
return {"sources": [h.metadata["source"] for h in hits]}
def retrieval_hit(run, example) -> dict:
return {"key": "retrieval_hit", "score": int(
example.outputs["source"] in run.outputs["sources"])}
evaluate(retrieve_target, data="客服黄金集",
evaluators=[retrieval_hit], experiment_prefix="cs-bot")
一次典型运行的控制台摘录如下：
8it [00:14, 1.74s/it]
Evaluation Report (data=客服黄金集): retrieval_hit 0.875
7/8 命中，未中的是"怎么导出看板数据"：export.md 的块排在第 5 位，k 取 4 天然够不着——与第 9 篇 checkpointer 那问同款悬案。把 search_kwargs 的 k 调到 6 重跑，第二轮实验 8/8 全中，代价是每问多喂两块上下文，token 与噪声同涨，这笔账由产品定夺，但底线一句话：分数不涨的改动不合入。答案层面的开放式验收交给 LLM as judge：手写 judge 用 with_structured_output 定义"理由加分数"的裁决结构，method 随集成包默认——OpenAI 走 json_schema，Anthropic 走 function_calling，第 9 篇第 5 节的写法原样适用。评测口径详见LangSmith 评估指南。
7、九篇能力对照
收官之际，把前九篇与本项目逐一对齐，各篇的落点如下：
| 篇目 | 本篇落点 |
|---|---|
| 第 1 篇：定位与第一次调用 | init_chat_model 一处声明，换模型只改前缀字符串 |
| 第 2 篇：提示模板 | 客服人设与护栏收进 system_prompt，单独可评审 |
| 第 3 篇：结构化输出 | 工单 args_schema 的字段描述与 Literal 枚举 |
| 第 4 篇：工具调用 | @tool 定义三工具，描述分工决定路由质量 |
| 第 5 篇：LCEL 链 | 检索器与 RunnableLambda 串成管道收进工具 |
| 第 6 篇：create_agent | 装配主体，stream 的 updates 模式做轨迹调试 |
| 第 7 篇：记忆与状态 | 双记忆装配：checkpointer 与 store 各管一层 |
| 第 8 篇：RAG 组件 | 切块嵌入入库，as_retriever 供检索与评测共用 |
| 第 9 篇：评测实战 | 黄金集命中率立基线，分数回归纪律 |
九个落点没有一处是复习，每个 API 都在生产位置干活；第 1 篇说框架的价值在第十次调用兑现，本篇就是那第十次——积木还是那九块，合在一起是一台能交付的机器。
8、扩展方向
两个方向接着本篇往外走。流程要更硬的控制时上 LangGraph：高优工单先人工审批再发出、按问题类型路由到不同处理图、多 agent 分工协作，create_agent 固定的双节点图不再够用。搭图的语法就是它内部那套模型节点、工具节点与条件边，《LangGraph 从入门到实战》系列逐篇拆开，工单审批中断正是其中的典型案例。
上线后补运行闭环：LangSmith 的 tracing 记录线上每轮对话的完整轨迹与 token 成本，从轨迹里挑 badcase 回流黄金集。第 9 篇的回归纪律由此从离线演练变成日常运转，评测闭环与监控闭环合成一张网。
9、总结与练习
回顾全文：收官项目把九块积木装进同一台机器——第 8 篇五步建知识库，第 4 篇装饰器与第 3 篇 schema 定义三个工具，第 5 篇管道收进检索工具，第 6、7 篇的 create_agent 加双记忆完成装配，三段对话各验一层能力，第 9 篇的黄金集命中率立住回归护栏。对于准备动手的第一个 agent 项目，建议照本篇顺序落地：知识库、工具、装配、主循环、评测五步各自可独立验证，出问题时定位到模块而不是玄学；内存实现跑通后按第 7 篇换落盘件，即可上生产。
练习两道：
- 加一个退订工具
cancel_subscription：首次调用只复述套餐与生效日期请用户确认，用户回复确认才执行，用两段式对话模拟审批 - 把双记忆换成落盘实现（
SqliteSaver与PostgresStore），重启进程后用原会话号复问验短期记忆，换新会话问偏好验长期记忆
