企业官网建设流程全解析

1. 这不是“又一篇Langchain教程”，而是我踩完所有坑后重写的实战手记

Langchain 官方使用指南笔记（五）——这个标题看起来平平无奇，甚至有点枯燥。但如果你正卡在 agent 编排逻辑混乱、Chain 调用链断裂、RAG 结果飘忽不定、或者刚把 LLM 接进系统就发现 prompt 一改全崩的阶段，那这篇笔记就是为你写的。它不讲“Langchain 是干嘛的”这种教科书定义，也不堆砌 API 列表，而是聚焦一个真实场景：如何让一个基于 Langchain 构建的生产级 RAG+Agent 混合系统，在模型切换、数据更新、用户追问、错误恢复等现实压力下，依然保持可预测、可调试、可维护的稳定输出。我用它落地过三个客户项目：某省政务知识助手（日均调用量 2.3 万+）、某医疗器械企业 SOP 智能检索系统（接入 17 类非结构化 PDF/扫描件）、以及一个面向开发者的内部技术文档问答平台（支持代码块提取与上下文补全）。这些项目共同暴露出 Langchain 官方文档里几乎不提的五个致命断点：状态持久化的隐式陷阱、Tool 调用失败时的默认熔断策略、Memory 在多轮对话中的 token 溢出静默截断、LLM 输出解析器（OutputParser）与自定义 Tool 返回格式的类型错配、以及最隐蔽的——Langchain 的 Runnable 接口在异步流式响应中对 chunk 边界处理的底层假设。这些不是“高级技巧”，而是你写完第一行from langchain_core.runnables import RunnablePassthrough就该知道的生存常识。本文所有代码、配置、参数值，全部来自这三套已上线系统的生产环境快照，连注释里的 warning 都是当时凌晨三点 debug 出来的原始日志片段。适合已经跑通pip install langchain并成功调用过一次ChatOpenAI的人，也适合正在评估 Langchain 是否该进入技术选型清单的架构师——因为我会告诉你，哪些模块你必须自己重写，哪些封装你最好绕开。

2. 核心设计思路：为什么放弃“标准 Agent 模板”，选择手动编排 Runnable 链

2.1 官方 Agent 模板的三大不可控性

Langchain 官方文档里反复推荐的create_react_agent或create_openai_functions_agent看似开箱即用，但在我部署的三个项目中，它们都成了后期运维的噩梦源头。问题不在于功能缺失，而在于其内部封装抹去了关键控制点：

• 状态管理黑盒化：官方 Agent 将agent_executor的RunnableConfig中的configurable字段与memory绑定，导致你无法在单次请求中动态注入用户会话 ID 或临时上下文。比如政务知识助手要求“同一市民 ID 的连续提问必须共享历史摘要”，但get_session_history回调函数只能返回BaseMessageHistory实例，而InMemoryChatMessageHistory的add_messages方法会无条件追加所有消息，包括系统生成的 intermediate steps。结果就是，当用户问“刚才说的补贴标准是多少？”，Agent 却从 5 轮前的对话中抓取了错误的数字。

• Tool 调用失败的静默降级：create_react_agent默认使用ReActSingleActionAgentOutputParser，当某个 Tool 执行超时或返回空结果时，它不会抛出异常，而是将tool_input原样塞回 LLM 的 next prompt，形成“LLM 替你猜答案”的恶性循环。我们在医疗器械项目中遇到过真实案例：一个查询“最新版 ISO 13485 条款”的 Tool 因向量数据库连接池耗尽返回空，Agent 却生成了一段看似专业实则虚构的条款描述，被法务部门直接叫停上线。

• 输出解析器与 Tool 返回值的强耦合陷阱：官方模板强制要求所有 Tool 的invoke方法返回str或dict，但实际业务中，Tool 往往需要返回结构化对象（如SearchResult类含score,source_id,page_num字段）。一旦你返回自定义类，JsonOutputParser会因TypeError: Object of type SearchResult is not JSON serializable直接崩溃，而文档里只字未提如何注册自定义序列化器。

提示：Langchain 的Runnable设计哲学是“一切皆可组合”，但官方 Agent 模板却反其道而行之，用一层厚重的胶水代码把LLM,Tools,Memory,OutputParser粘合成一个不可拆解的黑盒。这不是简化，而是把复杂性从代码层转移到了调试层。

2.2 我们的选择：用 RunnablePassthrough + RunnableParallel 手动构建“透明链”

我们彻底弃用了create_*_agent，转而用 Langchain Core 的原语（primitives）从零搭建执行链。核心思想是：让每一步的输入、输出、错误都暴露在日志里，让每一层的职责边界清晰到可以单独单元测试。以政务知识助手的典型流程为例：

1. 用户输入：“我孩子今年上小学，能申请哪些教育补贴？”
2. 首先，用RunnablePassthrough将原始输入透传给Router模块，根据关键词（“教育”、“补贴”、“小学”）路由到EducationSubsidyRetriever；
3. EducationSubsidyRetriever是一个Runnable子类，它内部调用Chroma向量库 +RecursiveCharacterTextSplitter分块器 +SentenceTransformerEmbeddings，并将 top-k 结果包装成Document列表；
4. 这些Document不直接喂给 LLM，而是先经过ContextualCompressor（一个自定义Runnable），它用轻量级 LLM（如Phi-3-mini）对每个Document的 relevance score 进行动态重排序，并过滤掉与“小学”无关的条款；
5. 最终，RunnableParallel将压缩后的上下文、用户原始问题、以及预设的 system prompt 并行组装成prompt_value，再交由ChatOpenAI处理；
6. LLM 输出后，PydanticOutputParser解析为SubsidyResponsePydantic 模型，其中eligibility_conditions字段自动校验是否包含“户籍”、“学籍”、“收入证明”等必填项。

这个链的每个环节都是独立的Runnable，你可以对ContextualCompressor单独压测其重排序准确率，也可以为EducationSubsidyRetriever注入 mock 的 Chroma client 来做离线测试。更重要的是，当某一步失败时，你能精准定位到是Chroma查询超时（捕获TimeoutError），还是Phi-3-mini重排序返回了空列表（检查len(compressed_docs) == 0），而不是在 Agent 的intermediate_steps日志里大海捞针。

2.3 为什么坚持用 Runnable 而非 Chain？

这里有个关键认知差：很多教程把Chain和Runnable当作同义词，但 Langchain v0.1+ 的演进方向非常明确——Chain是Runnable的语法糖，而Runnable是底层契约。Chain的run()方法是同步阻塞的，且其verbose日志只打印最终输入输出；而Runnable的invoke()/stream()/batch()方法提供了细粒度的生命周期钩子（on_start,on_error,on_end），这才是生产环境必需的可观测性基础。例如，我们在on_end钩子里记录每个Runnable的耗时、token 使用量、缓存命中率，并上报到 Prometheus。当发现ContextualCompressor的 P95 耗时突然从 120ms 升至 850ms，我们立刻知道是Phi-3-mini的 GPU 显存不足，而非笼统地认为“Agent 变慢了”。

3. 核心细节解析：五个必须亲手实现的关键模块

3.1 可审计的 Memory：解决多轮对话中的 token 溢出与上下文污染

Langchain 的ConversationBufferMemory看似简单，但它有一个致命缺陷：memory_key="chat_history"的load_memory_variables方法会无差别地将所有历史消息拼接成字符串，然后交给 LLM。当对话超过 20 轮，chat_history字符串轻松突破 4000 token，而你的 LLM context window 可能只有 8192。更糟的是，ConversationBufferMemory对SystemMessage和FunctionMessage的处理是统一的，导致 LLM 在生成回复时，会把上一轮的FunctionMessage（如{"result": "{'status': 'success', 'data': {...}}'}）当作普通对话内容来理解，从而产生幻觉。

我们的解决方案是：用ConversationSummaryBufferMemory替代，并强制其 summary_prompt 包含领域约束。具体实现如下：

from langchain.memory import ConversationSummaryBufferMemory from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import SystemMessage, HumanMessage, AIMessage # 定义一个带领域知识的摘要提示词 SUMMARY_PROMPT = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的政务知识助手摘要员。请严格遵循：\n" "1. 只总结用户明确提出的政策需求（如'补贴标准'、'申请流程'、'截止日期'）；\n" "2. 忽略所有闲聊、情绪表达、重复确认；\n" "3. 若用户提及具体文件名（如'粤教基〔2023〕12号'），必须保留在摘要中；\n" "4. 输出格式：纯文本，不超过 50 字，不带任何标点符号外的符号。"), MessagesPlaceholder(variable_name="messages") ]) # 初始化 memory，设置 max_token_limit 为 LLM context window 的 30% memory = ConversationSummaryBufferMemory( llm=ChatOpenAI(model="gpt-4-turbo", temperature=0), memory_key="chat_history", return_messages=True, max_token_limit=2400, # 8192 * 0.3 ≈ 2400 output_key="summary", # 关键！让 summary 成为可追踪的输出字段 input_key="input" # 明确指定输入字段名 )

这个ConversationSummaryBufferMemory的工作流程是：

• 每次save_context({"input": user_msg}, {"output": ai_msg})时，它先检查当前chat_history的总 token 数；
• 若超出max_token_limit，则触发SUMMARY_PROMPT，将最早的几轮对话压缩成一条SystemMessage，并替换掉原始历史；
• load_memory_variables返回的不再是原始消息列表，而是{"chat_history": [SystemMessage(content="补贴标准2023年调整为..."), HumanMessage(...), ...]}，其中第一条SystemMessage就是摘要。

注意：return_messages=True是必须的，否则load_memory_variables返回的是字符串，无法与MessagesPlaceholder兼容。而output_key="summary"让你在日志里能直接看到每次压缩生成的摘要内容，这是审计对话一致性的黄金字段。

3.2 可插拔的 Retriever：超越 Chroma 的混合检索策略

langchain_community.vectorstores.chroma是新手入门首选，但它的similarity_search_with_score方法返回的是(Document, score)元组，而生产环境需要的是：精确控制召回粒度、支持多源异构数据、能动态加权不同检索器的结果。我们构建了一个HybridRetriever，它同时运行Chroma（向量检索）、BM25Retriever（关键词检索）、以及一个SQLRetriever（结构化数据查询），最后用RRF（Reciprocal Rank Fusion）算法融合结果。

from langchain.retrievers import BM25Retriever from langchain_community.vectorstores import Chroma from langchain_community.utilities import SQLDatabase from langchain_core.documents import Document import numpy as np class HybridRetriever: def __init__(self, chroma_db: Chroma, sql_db: SQLDatabase): self.chroma_retriever = chroma_db.as_retriever(search_kwargs={"k": 5}) self.bm25_retriever = BM25Retriever.from_documents( documents=chroma_db._collection.get()["documents"], # 从 Chroma 获取原始文档 k=5 ) self.sql_retriever = SQLRetriever(sql_db) def invoke(self, query: str, **kwargs) -> List[Document]: # 并行执行三种检索 chroma_docs = self.chroma_retriever.invoke(query) bm25_docs = self.bm25_retriever.invoke(query) sql_docs = self.sql_retriever.invoke(f"SELECT * FROM policies WHERE title LIKE '%{query}%'") # RRF 融合：rank = 1/(rank_i + 60)，权重为 0.4, 0.3, 0.3 all_docs = chroma_docs + bm25_docs + sql_docs doc_to_scores = {} for i, doc in enumerate(chroma_docs): doc_to_scores[doc.page_content] = doc_to_scores.get(doc.page_content, 0) + 0.4 / (i + 60) for i, doc in enumerate(bm25_docs): doc_to_scores[doc.page_content] = doc_to_scores.get(doc.page_content, 0) + 0.3 / (i + 60) for i, doc in enumerate(sql_docs): doc_to_scores[doc.page_content] = doc_to_scores.get(doc.page_content, 0) + 0.3 / (i + 60) # 按 RRF score 降序排列，去重 sorted_docs = sorted( set(all_docs), key=lambda d: doc_to_scores.get(d.page_content, 0), reverse=True ) return sorted_docs[:5] # 返回 top-5

这个HybridRetriever的价值在于：当用户搜索“2024年广州小学入学政策”，Chroma可能召回《广州市义务教育招生工作细则（2023版）》，BM25精准匹配到《2024年招生日程安排表》的标题，而SQLRetriever则查出数据库中policies表里year=2024且city='Guangzhou'的最新政策 ID。RRF 算法确保即使某个检索器排名靠后，只要其他两个高度一致，该文档仍能获得高分。我们在政务项目中实测，混合检索将 top-1 准确率从纯向量的 68% 提升至 89%。

3.3 可验证的 OutputParser：用 Pydantic 强制结构化输出

Langchain 的JsonOutputParser在面对复杂嵌套结构时极其脆弱。例如，当 LLM 应该输出{"response": "符合资格", "conditions": [{"name": "户籍", "status": "满足"}, {"name": "学籍", "status": "待确认"}]}，但实际返回了{"response": "符合资格", "conditions": [{"name": "户籍", "status": "满足"}, {"name": "学籍"}]}（缺少status字段），JsonOutputParser会静默忽略整个conditions数组，而不是报错。这导致前端永远收不到conditions数据。

我们的方案是：用 Pydantic v2 的RootModel和model_validate_json构建零容忍解析器。

from pydantic import BaseModel, Field, RootModel from typing import List, Optional class Condition(BaseModel): name: str = Field(description="条件名称，如'户籍'、'学籍'") status: str = Field(description="状态，必须是'满足'、'不满足'、'待确认'之一") evidence: Optional[str] = Field(default=None, description="支持该状态的依据文本") class SubsidyResponse(BaseModel): response: str = Field(description="最终结论，如'符合资格'、'不符合资格'") conditions: List[Condition] = Field(description="所有核查条件的状态列表") source_references: List[str] = Field(description="引用的政策文件编号列表，如['粤教基〔2023〕12号']") # 自定义 Parser，继承 Runnable class StrictPydanticOutputParser(Runnable): model: Type[BaseModel] def invoke(self, input: str, config: Optional[RunnableConfig] = None) -> BaseModel: try: # 强制要求 JSON 格式，且必须包含所有 required 字段 return self.model.model_validate_json(input) except Exception as e: # 记录原始 LLM 输出和错误，用于后续 prompt 优化 logger.error(f"StrictParse failed for {input[:100]}... Error: {e}") raise ValueError(f"LLM output invalid for {self.model.__name__}: {e}") # 使用方式 parser = StrictPydanticOutputParser(model=SubsidyResponse)

这个StrictPydanticOutputParser的invoke方法会严格校验：

• input是否为合法 JSON；
• 所有Field(...)标记的字段是否都存在；
• status字段的值是否在枚举范围内；
• source_references是否为字符串列表。

一旦校验失败，它会抛出带上下文的ValueError，触发上游的RetryPolicy（我们为每个Runnable配置了指数退避重试），而不是让错误数据流入下游。这比任何try...except都更可靠，因为它是 schema 层面的强制约束。

3.4 可观测的 Tool：为每个 Tool 注入 trace_id 与性能指标

Langchain 的Tool类本身不提供监控能力。当EducationSubsidyRetriever调用Chroma超时时，你只能看到ToolException，却不知道是网络抖动、Chroma server OOM，还是查询向量维度不匹配。我们的做法是：将每个 Tool 封装为Runnable，并在invoke中注入 OpenTelemetry trace。

from opentelemetry import trace from opentelemetry.trace import Status, StatusCode class TracedTool(Runnable): def __init__(self, tool_func: Callable, name: str): self.tool_func = tool_func self.name = name self.tracer = trace.get_tracer(__name__) def invoke(self, input: Dict, config: Optional[RunnableConfig] = None) -> Any: with self.tracer.start_as_current_span(f"tool.{self.name}") as span: try: # 记录输入参数（脱敏后） span.set_attribute("tool.input.keys", list(input.keys())) if "query" in input: span.set_attribute("tool.input.query_length", len(input["query"])) result = self.tool_func(input) # 记录输出长度与类型 span.set_attribute("tool.output.type", type(result).__name__) if isinstance(result, (str, list)): span.set_attribute("tool.output.length", len(str(result))) span.set_status(Status(StatusCode.OK)) return result except Exception as e: span.set_status(Status(StatusCode.ERROR, str(e))) span.record_exception(e) raise e # 使用 education_tool = TracedTool( tool_func=lambda x: hybrid_retriever.invoke(x["query"]), name="education_subsidy_retriever" )

这样，当education_tool被调用时，Jaeger 或 Grafana Tempo 就能显示完整的 trace：从tool.education_subsidy_retriever的耗时、错误率，一直下钻到chroma.search的 P99 延迟。我们在上线前用此 trace 发现了 Chroma 的hnsw索引在高并发下内存泄漏的问题，提前规避了线上事故。

3.5 可恢复的 AgentExecutor：手动实现 ReAct 循环的熔断与重试

既然放弃了官方AgentExecutor，我们就得自己写一个。核心是：将 ReAct 的“思考-行动-观察”循环拆解为可中断、可重试、可审计的步骤。

from langchain_core.runnables import RunnableLambda from typing import Dict, Any, List, Optional class CustomAgentExecutor: def __init__(self, llm: ChatOpenAI, tools: List[Runnable], max_iterations: int = 5): self.llm = llm self.tools = {tool.name: tool for tool in tools} self.max_iterations = max_iterations def invoke(self, input: Dict[str, Any], config: Optional[RunnableConfig] = None) -> Dict[str, Any]: # 初始化中间状态 state = { "input": input["input"], "chat_history": input.get("chat_history", []), "intermediate_steps": [], "iteration": 0 } while state["iteration"] < self.max_iterations: # Step 1: LLM 思考，生成 Action action_prompt = self._build_action_prompt(state) action_output = self.llm.invoke(action_prompt) # Step 2: 解析 Action（这里用正则，比官方 parser 更鲁棒） action_match = re.search(r"Action: ([^\n]+)\nAction Input: (.+)", action_output.content, re.DOTALL) if not action_match: # 未识别出 Action，视为最终回答 return {"output": action_output.content} tool_name, tool_input = action_match.groups() if tool_name not in self.tools: # 工具不存在，记录错误并重试 state["intermediate_steps"].append({ "error": f"Unknown tool: {tool_name}", "retry_count": 0 }) state["iteration"] += 1 continue # Step 3: 执行 Tool，带重试 try: tool_result = self._execute_tool_with_retry( self.tools[tool_name], {"query": tool_input.strip('"')}, max_retries=2 ) except Exception as e: # Tool 执行失败，记录并重试 state["intermediate_steps"].append({ "tool": tool_name, "input": tool_input, "error": str(e), "retry_count": 0 }) state["iteration"] += 1 continue # Step 4: 观察结果，加入历史 state["intermediate_steps"].append({ "tool": tool_name, "input": tool_input, "output": str(tool_result)[:500] # 截断避免 token 爆炸 }) state["iteration"] += 1 # 达到最大迭代次数，强制返回最终思考 final_prompt = self._build_final_prompt(state) final_output = self.llm.invoke(final_prompt) return {"output": final_output.content, "intermediate_steps": state["intermediate_steps"]} def _execute_tool_with_retry(self, tool: Runnable, input: Dict, max_retries: int) -> Any: for i in range(max_retries + 1): try: return tool.invoke(input) except Exception as e: if i == max_retries: raise e time.sleep(0.5 * (2 ** i)) # 指数退避

这个CustomAgentExecutor的优势在于：

• intermediate_steps是一个结构化列表，每一步都包含tool,input,output,error，可直接存入 Elasticsearch 供审计；
• max_iterations和max_retries是硬性熔断开关，防止无限循环；
• _execute_tool_with_retry内置指数退避，比官方ToolException的静默处理更可控。

4. 实操过程：从零搭建一个可上线的 RAG+Agent 混合系统

4.1 环境准备与依赖锁定

不要用pip install langchain这种模糊命令。生产环境必须锁定所有依赖版本，尤其是langchain-core和langchain-community，因为它们的 API 在小版本间常有破坏性变更。我们使用pyproject.toml精确声明：

[tool.poetry.dependencies] python = "^3.10" langchain-core = "0.1.49" langchain-community = "0.0.34" langchain-openai = "0.1.12" chromadb = "0.4.24" sentence-transformers = "2.7.0" openai = "1.35.1" pydantic = {version = "2.7.1", extras = ["dotenv"]} opentelemetry-api = "1.24.0" opentelemetry-sdk = "1.24.0"

注意：langchain-core==0.1.49是关键。0.1.50 版本引入了RunnableBinding的新行为，会导致我们自定义的TracedTool的invoke方法签名不兼容。这个版本号是我们三个项目经过两周灰度验证后确定的稳定基线。

4.2 数据预处理：PDF 解析与分块的工业级实践

RAG 效果 70% 取决于数据质量。PyPDFLoader对扫描版 PDF 完全无效，UnstructuredPDFLoader在中文长文档中常出现乱码。我们采用分层解析策略：

1. 第一层：OCR 识别（针对扫描件）

   • 使用paddleocr（比 Tesseract 中文识别准确率高 23%）对 PDF 每页进行 OCR；
   • 将 OCR 结果保存为page_{i}_ocr.txt，并记录page_num,confidence。

2. 第二层：结构化解析（针对文字版 PDF）

   • 用pdfplumber提取每页的文本、表格、字体大小；
   • 根据字体大小变化（如标题用 16pt，正文用 12pt）和空白行，自动识别章节结构；
   • 将每个“逻辑段落”（如一个条款、一个问答）作为独立Document。

3. 第三层：智能分块

   • 不用RecursiveCharacterTextSplitter的固定chunk_size=1000，而是用SemanticChunker（基于all-MiniLM-L6-v2）：

     from langchain_text_splitters import SemanticChunker from langchain_community.embeddings import HuggingFaceEmbeddings embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2") text_splitter = SemanticChunker( embeddings, breakpoint_threshold_type="percentile", # 按语义相似度百分位切分 breakpoint_threshold_amount=95 # 只在相似度低于 95% 分位处切分 )

最终，一份 50 页的《广东省义务教育条例》PDF，会被解析为 127 个语义连贯的Document，每个Document的metadata包含source_file,page_range,section_title,ocr_confidence。这比简单分块的召回准确率高出 41%。

4.3 向量库部署：Chroma 的生产级配置

Chroma 的默认配置（PersistentClient+DiskStorage）在高并发下极易出现sqlite3.DatabaseError: database is locked。我们改为：

• 服务端模式：启动chroma run --host 0.0.0.0 --port 8000，用HttpClient连接；
• 索引优化：创建 collection 时指定hnsw参数：

  collection = client.create_collection( name="policy_docs", metadata={ "hnsw:space": "cosine", # 余弦相似度 "hnsw:construction_ef": 128, # 构建时邻居数 "hnsw:M": 64 # 每个节点的最大连接数 } )

• 批量插入：不用add()单条插入，而用upsert()批量：

  # 每批 100 条，避免内存溢出 for i in range(0, len(docs), 100): batch_docs = docs[i:i+100] collection.upsert( ids=[f"doc_{j}" for j in range(i, i+len(batch_docs))], documents=[doc.page_content for doc in batch_docs], metadatas=[doc.metadata for doc in batch_docs] )

4.4 LLM 选型与 Prompt 工程：GPT-4-turbo 与本地模型的协同

我们不迷信单一模型。生产环境采用“双模驱动”：

• 主模型：gpt-4-turbo，负责最终回答生成、摘要、复杂推理；
• 辅模型：Phi-3-mini（4K context），部署在本地 GPU 上，负责ContextualCompressor、BM25重排序、OutputParser的预校验。

Prompt 设计上，我们摒弃了“角色扮演”式提示，改用指令式（Instruction-based）Prompt，因为它对模型微调更友好：

你是一个政务知识助手，任务是根据提供的政策文档和用户问题，生成准确、简洁、可验证的回答。 【约束】 - 必须严格基于文档内容，禁止编造、推测、添加个人意见； - 若文档未提及某信息，回答“未找到相关信息”； - 所有政策依据必须标注文件编号，如“依据《粤教基〔2023〕12号》第3条”； - 回答长度不超过 200 字。 【输入】 用户问题：{input} 政策文档：{context} 【输出】

这个 Prompt 在 GPT-4-turbo 上的“依据标注准确率”达到 98.7%，远高于传统角色提示的 82.3%。

4.5 部署与监控：FastAPI + Prometheus + Grafana

后端用 FastAPI 封装CustomAgentExecutor：

from fastapi import FastAPI, HTTPException from pydantic import BaseModel app = FastAPI() class QueryRequest(BaseModel): input: str session_id: str @app.post("/ask") async def ask(request: QueryRequest): try: # 注入 session_id 到 memory config = {"configurable": {"session_id": request.session_id}} result = agent_executor.invoke( {"input": request.input}, config=config ) return {"output": result["output"]} except Exception as e: logger.exception("Ask failed") raise HTTPException(status_code=500, detail=str(e))

监控指标：

• langchain_runnable_duration_seconds：每个Runnable的耗时直方图；
• langchain_tool_call_total：按tool_name统计的调用次数；
• langchain_parser_failure_total：StrictPydanticOutputParser的失败次数；
• chroma_query_duration_seconds：Chroma 查询延迟。

Grafana 看板中，我们设置了关键告警：

• rate(langchain_parser_failure_total[1h]) > 5：1 小时内解析失败超 5 次，说明 prompt 或模型需优化；
• histogram_quantile(0.95, rate(langchain_runnable_duration_seconds_bucket[1h])) > 5：P95 耗时超 5 秒，需检查 LLM 或工具瓶颈。

5. 常见问题与排查技巧实录：那些文档里绝不会写的真相

5.1 “我的 RAG 总是答非所问，但向量相似度分数很高”——真相是：相似度分数是相对的，不是绝对的

Chroma 的similarity_search_with_score返回的score是余弦相似度，范围是 [-1, 1]。但新手常误以为score > 0.8就是“高度相关”。实际上，这个分数只在同一查询、同一 collection 内部才有比较意义。我们曾遇到一个案例：用户问“补贴标准”，Chroma 返回了 5 个score在 0.72~0.78 的文档，但其中 4 个是关于“创业补贴”的，只有 1 个是“教育补贴”。根本原因在于，all-MiniLM-L6-v2这类通用 embedding 模型，对“补贴”一词的向量表示过于宽泛，无法区分“教育”、“创业”、“住房”等修饰词。解决方案是：

• 领域微调：用政务政策语料对all-MiniLM-L6-v2进行 LoRA 微调，重点强化“教育补贴”、“创业补贴”等短语的区分度；
• 查询重写：在检索前，用 LLM 将用户问题重写为“教育补贴 标准 2024 广州”，增加关键词密度；
• MMR 重排序：用Maximum Marginal Relevance算法，在保证相关性的同时，提升结果多样性。

5.2 “Agent 执行 Tool 后，LLM 总是忽略观察结果”——真相是：LLM 的 prompt template 没有给 Tool 输出留出足够空间

官方ReActprompt 模板中，“Observation: ”后面直接跟 Tool 输出，但当 Tool 返回一个 2000 字的政策原文时，LLM 的 context window 就被占满，导致它没“看到”自己的 Action。我们的 fix 是：

• 截断 + 摘要：CustomAgentExecutor中，tool_result在加入intermediate_steps前，先用Phi-3-mini生成 50 字摘要；
• 显式分隔符：在 prompt 中用---OBSERVATION START---和---OBSERVATION END---包裹摘要，强化 LLM 对 Observation 的识别。

5.3 “为什么同样的代码，在 Jupyter 里