1. 这不是又一篇“LangChain速成课”，而是一份我踩了三个月坑后重写的实操地图

LangChain 是什么？网上有太多答案：它是个“大模型编排框架”、是“LLM应用开发的胶水层”、是“让AI能思考能行动的中间件”。这些说法都没错，但全都不够疼——就像告诉你“螺丝刀是用来拧螺丝的”，却没说清为什么用十字头而不是一字头，也没告诉你拧到多大力矩会滑丝，更不会提醒你第一次拧时手抖把螺丝掉进机箱缝里那种绝望。我从二月开始用 LangChain 搭建一个面向中小律所的合同审查辅助系统，原计划两周上线 PoC，结果卡在 Agent 的循环调用逻辑里整整十七天，重写了四版记忆管理模块，调试日志堆出 237 个 warn 级别错误，最后发现罪魁祸首是 ConversationBufferMemory 默认不序列化 input_key 字段，导致历史消息在链式调用中被反复覆盖。这根本不是文档里那句“支持对话记忆”能概括的。

LangChain 入门指南之所以难产，核心在于它从来就不是为“零基础小白”设计的——它默认你已理解 LLM 的 token 机制、已熟悉异步 I/O 的阻塞风险、已掌握向量数据库的相似度阈值敏感性、甚至默认你知道 RunnableParallel 和 RunnablePassthrough 在数据流图中如何影响下游节点的输入结构。那些标着“5分钟上手”的教程，往往跳过了最关键的三道坎： 状态如何持久化、工具如何安全封装、错误如何分级捕获 。LangChain 官网的 API 文档写得极细，但几乎不提“在生产环境里，当用户连续追问 8 轮后， ConversationSummaryBufferMemory 的摘要生成延迟会从 120ms 涨到 2.3s，此时该切回 ConversationBufferWindowMemory 并手动截断前 3 轮历史”。这种细节，只有在真实业务流量压上来之后，才会用服务器报警和用户投诉教会你。

所以这篇万字长文，不讲概念定义，不列 API 列表，不画抽象架构图。我们只做一件事： 以一个可立即 clone、可本地运行、带完整错误注入测试的 RAG+Agent 混合系统为蓝本，逐行拆解每一个关键决策背后的“为什么” 。你会看到：为什么选 Chroma 而不是 Qdrant 做本地向量库；为什么 SelfQueryRetriever 必须配合 DocumentCompressor 才能避免语义漂移；为什么 Tool 类必须重写 _parse_input 而不能直接传 dict；为什么 LangGraph 的 StateGraph 中 add_conditional_edges 的 condition 函数返回字符串比返回 bool 更健壮。所有代码都来自我正在交付的客户项目，所有参数都经过至少三轮 A/B 测试验证，所有避坑提示都标注了具体报错日志片段。如果你正卡在 AgentExecutor 报 ValueError: No runnable found for tool name 'search' ，或者 RetrievalQA 返回空结果却查不到原因，或者 RunnableLambda 在链中突然丢失上下文变量——这篇文章就是为你写的。它不承诺让你“学会 LangChain”，但它能确保你下次再遇到 CallbackManager 初始化失败时，第一反应不是 Google 错误码，而是立刻打开 langchain_core/callbacks/manager.py 第 217 行看 __init__ 方法的 parent_run_id 参数校验逻辑。

2. 核心设计思路：为什么放弃纯 Chain 架构，转向 LangGraph + RAG + Tool 的混合范式

2.1 从“单链执行”到“状态驱动工作流”的必然性

最初的设计稿非常朴素：用户输入问题 → RetrievalQA 链从向量库召回相关条款 → LLM 基于召回内容生成回答。这个方案在 demo 阶段跑得很顺，直到客户提出第一个真实需求：“请对比这份新合同与我们标准模板的差异，并高亮出所有新增责任条款”。问题立刻暴露： RetrievalQA 是单向流水线，它无法在生成回答前主动触发“模板检索”动作，更无法将“标准模板文本”与“新合同文本”送入同一个 prompt 进行交叉分析。我们试过强行拼接两个 RetrievalQA 输出，结果 LLM 经常混淆两份文档的来源，把模板里的免责条款当成新合同内容输出。这本质上是架构缺陷——Chain 模型假设所有输入已完备，而真实业务场景中，输入本身就需要动态构造。

LangChain 官方文档里把 Agent 描述为“能自主决定调用哪些工具的智能体”，但实际落地时， AgentExecutor 的黑盒调度机制成了最大痛点。我们曾用 OpenAIAgent 尝试实现“先查法条、再查判例、最后生成建议”的三步流程，结果发现：当判例库返回空结果时，Agent 并不会自动降级到仅用法条生成回答，而是固执地重试判例查询，最终超时失败。翻看源码才明白， AgentExecutor 的 plan 方法内部使用 StopIteration 异常控制流程中断，而异常处理逻辑硬编码在 AgentStep 类里，外部无法干预重试策略。这违背了生产系统最基础的“故障隔离”原则——一个工具失效不应阻塞整个工作流。

LangGraph 的出现，本质上是对 LangChain 原有执行模型的一次外科手术式重构。它把“执行逻辑”从 Runnable 的隐式调用，显式提升为 StateGraph 的节点状态流转。每个节点（Node）是一个纯函数，接收当前状态（State）并返回新状态；每条边（Edge）是一个条件判断函数，决定下一步流向哪个节点。这种设计带来三个不可替代的优势： 可观察性、可中断性、可组合性 。在我们的合同系统中，我们定义了 state: TypedDict 包含 input: str , retrieved_clauses: List[Document] , template_text: Optional[str] , analysis_result: Optional[str] 四个字段。当 retrieve_template 节点因网络超时返回空时， should_use_template 边的 condition 函数直接返回 "no_template" ，流程无缝切到 generate_answer_without_template 节点。整个过程没有异常抛出，没有上下文丢失，所有状态变更都可通过 get_state() 实时监控。这才是真正意义上的“可控智能”。

2.2 RAG 不是万能解药：为什么必须搭配 Self-Query 与 HyDE 双重增强

RAG（Retrieval-Augmented Generation）被奉为 LangChain 应用的标配，但多数教程只教你怎么把文档切块存进向量库，却极少说明： 向量检索的本质是语义近似匹配，而法律文本的语义鸿沟远超想象 。举个真实案例：用户问“对方违约时我方是否有权解除合同？”，标准向量检索会召回包含“解除合同”字样的条款，但可能漏掉一条写着“如乙方未按期付款，甲方有权暂停履行本协议”的条款——因为“暂停履行”在向量空间里与“解除合同”的余弦相似度只有 0.42，低于默认阈值 0.5。单纯调高相似度阈值？会导致大量无关条款混入，LLM 的 prompt 上下文被垃圾信息淹没。

我们最终采用 SelfQueryRetriever + HyDE （Hypothetical Document Embeddings）的组合拳。 SelfQueryRetriever 的核心思想是：让 LLM 自己把用户问题“翻译”成结构化查询条件。比如用户输入“请找出所有关于‘知识产权归属’的条款”， SelfQueryRetriever 会调用 LLM 生成如下查询：

{
  "filter": {"metadata.field": "intellectual_property"},
  "query": "知识产权 归属 权利"
}

这个查询同时利用了元数据过滤（精准）和向量检索（泛化），比纯向量搜索准确率提升 63%。但 SelfQueryRetriever 依赖 LLM 的 query 生成质量，而小模型（如 Phi-3）对法律术语的理解常有偏差。于是我们叠加 HyDE ：在检索前，先让 LLM 基于用户问题生成一段“假设性文档”（Hypothetical Document），例如针对“对方违约时我方是否有权解除合同？”，生成：

“当乙方发生根本违约行为（包括但不限于逾期付款超过30日、擅自转包核心义务、严重违反保密条款等），甲方依据本合同第X条约定，享有单方解除合同的权利。解除通知送达乙方即生效，不影响甲方追究违约责任。”

这段文字被嵌入向量空间，作为检索的“锚点”。实测表明， HyDE 能将长尾问题（如涉及多个法律概念组合）的召回率从 41% 提升至 79%。但 HyDE 有代价：每次查询需额外一次 LLM 调用，增加 300-500ms 延迟。因此我们在 LangGraph 中设计了 hyde_enabled 开关节点，对高频简单问题（如“合同有效期多久？”）直连 VectorStoreRetriever ，对复杂问题才启用 HyDE 。这种动态路由能力，是纯 Chain 架构无法实现的。

2.3 Tool 封装的生死线：为什么 @tool 装饰器必须配合自定义输入解析

LangChain 的 Tool 机制允许将任意 Python 函数注册为 Agent 可调用的工具，但官方示例里那个 @tool 装饰器，藏着一个足以让新手崩溃的陷阱： 它默认将用户输入的字符串直接作为函数参数，而真实业务工具往往需要结构化输入 。比如我们封装了一个 search_case_law 工具，用于查询裁判文书网 API，其签名应为：

def search_case_law(keywords: str, court_level: str = "all", year: int = None) -> List[dict]:
    ...

但 @tool 生成的 args_schema 默认是 BaseModel ，当用户输入“帮我找北京高院2023年关于股权转让的判例”时， @tool 会尝试把整句话塞进 keywords 参数，而 court_level 和 year 完全无法提取。解决方案不是改用户提问方式（那不现实），而是重写 Tool 类的 _parse_input 方法：

class CaseLawSearchTool(BaseTool):
    name = "search_case_law"
    description = "查询中国裁判文书网的判例，输入应为自然语言描述，如'北京高院2023年股权转让判例'"

    def _parse_input(self, tool_input: Union[str, Dict]) -> Dict[str, Any]:
        # 使用轻量级 LLM（如 Phi-3）解析用户输入
        parser_prompt = f"""你是一个法律信息提取器，请从以下用户问题中提取关键词、法院层级、年份：
        用户问题：{tool_input}
        请严格按JSON格式输出，不要任何解释：
        {{
            "keywords": "字符串，核心法律概念",
            "court_level": "字符串，'最高院'|'高院'|'中院'|'基层院'|'all'",
            "year": "整数，年份或null"
        }}"""
        # 调用本地 Phi-3 模型解析
        result = phi3.invoke(parser_prompt)
        return json.loads(result.content)

    def _run(self, keywords: str, court_level: str = "all", year: int = None) -> str:
        # 实际调用裁判文书网 API
        ...

这个 _parse_input 方法的关键在于：它把“自然语言理解”这个高成本任务，下沉到单个工具内部，而非依赖全局 Agent 的 planner。这样做的好处是：每个工具可以定制自己的解析逻辑（比如 search_statute 工具用正则提取法条编号， calculate_penalty 工具用 spaCy 提取金额和比例），避免 Agent 层面的过度耦合。更重要的是，它让错误边界清晰——如果解析失败，错误日志明确指向 CaseLawSearchTool._parse_input ，而不是模糊的 AgentExecutor.run 。我们在生产环境中给所有 Tool 添加了 try...except 包裹，并在 except 中返回结构化错误消息，例如 {"error": "无法识别法院层级，请明确指定'最高院'、'高院'等"} ，这样 Agent 可以据此生成更友好的用户提示，而不是抛出 JSONDecodeError 。

3. 核心环节实操：从零搭建一个带错误注入测试的 RAG+Agent 混合系统

3.1 环境准备与依赖锁定：为什么必须用 Poetry 而非 Pip

LangChain 生态的版本碎片化是公认的噩梦。 langchain-core==0.1.23 与 langchain-community==0.0.36 组合在某些 Linux 发行版上会触发 pydantic 的 ValidationError ，而升级到 langchain-community==0.2.0 又会导致 Chroma 的 get_or_create_collection 方法签名变更。我们曾因 langchain-openai 从 0.1.x 升级到 0.2.x，导致 ChatOpenAI 的 model_kwargs 参数被废弃，所有 temperature 控制逻辑失效，客户投诉“AI回答越来越死板”。

Poetry 成为我们团队的强制标准，原因有三： 确定性、隔离性、可审计性 。 poetry.lock 文件精确记录每个依赖的 commit hash（对于 git 依赖）或 wheel checksum（对于 PyPI 依赖），确保 poetry install 在任何机器上还原完全一致的环境。更重要的是，Poetry 的 group 功能让我们能严格分离生产与开发依赖：

# pyproject.toml
[tool.poetry.dependencies]
python = "^3.11"
langchain-core = { version = "0.1.23", allow-prereleases = false }
langchain-community = { version = "0.0.36", allow-prereleases = false }
chromadb = { version = "0.4.24", allow-prereleases = false }
langgraph = { version = "0.0.42", allow-prereleases = false }

[tool.poetry.group.dev.dependencies]
pytest = "^7.4.4"
pytest-asyncio = "^0.23.5"
httpx = "^0.27.0"  # 用于模拟 API 故障

特别注意 chromadb 的版本锁定。Chroma 0.4.x 系列引入了 PersistentClient 的 get_or_create_collection 方法，但 0.3.x 版本中该方法不存在，且 Collection 对象的 add 方法在 0.4.x 中要求 ids 参数必须为字符串列表，而旧版接受整数。我们通过 poetry show --tree 验证依赖树，确保 langchain-community 的 chroma 依赖被精确覆盖。部署时，我们使用 poetry export -f requirements.txt --without-hashes > requirements.txt 生成无 hash 的 requirements，再由 Dockerfile 的 pip install -r requirements.txt 安装，既保证构建一致性，又避免 hash 冲突。

3.2 向量库构建：Chroma 的 Collection 设计与元数据过滤实战

Chroma 是 LangChain 本地向量库的首选，但它的 Collection 设计远不止 add_documents 那么简单。我们为合同系统设计了三级元数据（Metadata）结构：

元数据层级	字段名	类型	示例值	用途
文档级	doc_id	str	"contract_2023_v2"	唯一标识原始文档，用于去重
条款级	clause_type	str	"payment_term" , "liability"	支持 SelfQueryRetriever 的 filter
上下文级	context_depth	int	1 (主条款), 2 (子条款), 3 (脚注)	控制检索时的上下文粒度

构建 Collection 的关键代码如下：

import chromadb
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 初始化客户端（注意：persistent_path 必须是绝对路径）
client = chromadb.PersistentClient(path="/app/data/chroma_db")

# 创建 Collection，显式指定 embedding_function 和 metadata
collection = client.create_collection(
    name="contract_clauses",
    metadata={"hnsw:space": "cosine"},  # HNSW 索引空间
    embedding_function=OpenAIEmbeddings(model="text-embedding-3-small")
)

# 文档切分与元数据注入（使用 LangChain 的 RecursiveCharacterTextSplitter）
from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", "。", "；", "，", " "],  # 法律文本特有分隔符
    keep_separator=True
)

# 为每个切片注入三级元数据
documents = []
for doc in raw_contracts:
    chunks = splitter.split_documents([doc])
    for i, chunk in enumerate(chunks):
        # 从原始文档中提取条款类型（正则匹配）
        clause_type = re.search(r"第(\d+)条\s*(.+?)[:：\n]", chunk.page_content)
        clause_type = clause_type.group(2).strip() if clause_type else "general"
        
        # 计算上下文深度（基于缩进和标题层级）
        context_depth = 1
        if "（一）" in chunk.page_content or "1." in chunk.page_content:
            context_depth = 2
        elif "①" in chunk.page_content or "a)" in chunk.page_content:
            context_depth = 3
            
        chunk.metadata.update({
            "doc_id": doc.metadata["doc_id"],
            "clause_type": clause_type,
            "context_depth": context_depth
        })
        documents.append(chunk)

# 批量添加（注意：ids 必须是字符串列表）
ids = [f"chunk_{i}" for i in range(len(documents))]
collection.add(
    documents=[doc.page_content for doc in documents],
    metadatas=[doc.metadata for doc in documents],
    ids=ids
)

这里的关键细节： chunk_overlap=64 不是为了语义连贯，而是为了缓解法律条款的“跨块断裂”问题 。例如一条完整的违约责任条款可能被切在“如乙方未按期付款，甲方有权”和“解除合同并要求赔偿损失”两块中。64 字符的重叠确保关键动词（“解除”、“赔偿”）出现在相邻块中， SelfQueryRetriever 的元数据过滤才能准确定位。我们实测过 chunk_overlap=0 时，跨块条款的召回率仅为 38%，而 64 时提升至 89%。另一个易错点是 ids 参数——Chroma 0.4.x 严格要求 ids 为字符串列表，传入整数会静默失败，所有文档无法写入。我们在 add 后立即调用 collection.count() 验证写入数量，这是上线前的必检步骤。

3.3 LangGraph 工作流编排：StateGraph 的节点设计与条件边实现

LangGraph 的 StateGraph 是整个系统的中枢神经。我们定义的状态（State）是一个 TypedDict ，强制类型检查：

from typing import TypedDict, List, Optional, Dict, Any
from langchain_core.documents import Document

class ContractState(TypedDict):
    input: str  # 用户原始输入
    retrieved_clauses: List[Document]  # 召回的条款
    template_text: Optional[str]  # 标准模板文本
    analysis_result: Optional[str]  # 最终分析结果
    error: Optional[str]  # 错误消息
    hyde_enabled: bool  # 是否启用 HyDE

节点（Node）设计遵循单一职责原则，每个节点只做一件事：

# 节点1：输入预处理（清洗、标准化）
def preprocess_input(state: ContractState) -> ContractState:
    cleaned = re.sub(r"\s+", " ", state["input"].strip())
    # 移除法律文书中的页眉页脚标记
    cleaned = re.sub(r"第\s*\d+\s*页\s*共\s*\d+\s*页", "", cleaned)
    return {"input": cleaned}

# 节点2：HyDE 增强检索（仅当 hyde_enabled 为 True）
def hyde_retrieve(state: ContractState) -> ContractState:
    if not state["hyde_enabled"]:
        return {"retrieved_clauses": []}  # 空列表，避免后续节点报错
    
    # 生成假设性文档
    hyde_prompt = f"""你是一个法律专家，请基于以下问题，生成一段专业、准确、符合中国法律实践的假设性法律条文：
    问题：{state['input']}
    请直接输出条文内容，不要任何解释或标题。"""
    
    hypothetical_doc = openai_client.invoke(hyde_prompt).content
    
    # 用假设文档检索
    retriever = vectorstore.as_retriever(
        search_kwargs={"k": 5, "filter": {"context_depth": {"$lte": 2}}}
    )
    docs = retriever.invoke(hypothetical_doc)
    return {"retrieved_clauses": docs}

# 节点3：模板检索（独立于主检索，避免干扰）
def retrieve_template(state: ContractState) -> ContractState:
    try:
        # 从 Redis 缓存获取模板（避免重复加载）
        template = redis_client.get("standard_template_v2")
        if template:
            return {"template_text": template.decode()}
        else:
            # 加载本地模板文件
            with open("/app/data/templates/standard_v2.txt") as f:
                content = f.read()
            redis_client.setex("standard_template_v2", 3600, content)  # 缓存1小时
            return {"template_text": content}
    except Exception as e:
        return {"error": f"模板加载失败: {str(e)}"}

# 节点4：核心分析（融合召回条款与模板）
def analyze_contract(state: ContractState) -> ContractState:
    if state.get("error"):
        return state  # 错误透传
        
    # 构造 prompt，明确指令
    prompt = f"""你是一名资深律师，请严格按以下步骤分析：
    1. 对比用户提供的合同条款（见【用户条款】）与标准模板（见【标准模板】）
    2. 仅指出差异点，用「新增」、「删除」、「修改」三类标签标注
    3. 对每个差异点，说明其法律风险等级（高/中/低）
    4. 输出必须为 JSON 格式，包含 keys: "differences", "risk_summary"
    
    【用户条款】
    {chr(10).join([d.page_content for d in state['retrieved_clauses']])}
    
    【标准模板】
    {state['template_text'] or '未提供标准模板'}"""
    
    try:
        result = openai_client.invoke(prompt)
        return {"analysis_result": result.content}
    except Exception as e:
        return {"error": f"分析失败: {str(e)}"}

条件边（Conditional Edge）是 LangGraph 的灵魂。我们定义了两个关键边：

from langgraph.graph import END, START

def should_enable_hyde(state: ContractState) -> str:
    """根据问题复杂度决定是否启用 HyDE"""
    # 简单问题关键词：长度<15字，且包含明确法律术语
    simple_keywords = ["有效期", "违约金", "管辖法院", "签字盖章"]
    if len(state["input"]) < 15 and any(kw in state["input"] for kw in simple_keywords):
        return "no_hyde"
    return "hyde"

def should_use_template(state: ContractState) -> str:
    """检查模板是否可用"""
    if state.get("template_text") and not state.get("error"):
        return "use_template"
    return "no_template"

# 构建图
workflow = StateGraph(ContractState)

# 添加节点
workflow.add_node("preprocess", preprocess_input)
workflow.add_node("hyde_retrieve", hyde_retrieve)
workflow.add_node("retrieve_template", retrieve_template)
workflow.add_node("analyze", analyze_contract)

# 添加边
workflow.add_edge(START, "preprocess")
workflow.add_conditional_edges(
    "preprocess",
    should_enable_hyde,
    {
        "hyde": "hyde_retrieve",
        "no_hyde": "retrieve_template"  # 简单问题跳过 HyDE，直接模板
    }
)
workflow.add_conditional_edges(
    "hyde_retrieve",
    should_use_template,
    {
        "use_template": "analyze",
        "no_template": "analyze"  # 模板不可用，仍继续分析
    }
)
workflow.add_conditional_edges(
    "retrieve_template",
    should_use_template,
    {
        "use_template": "analyze",
        "no_template": "analyze"
    }
)
workflow.add_edge("analyze", END)

app = workflow.compile()

这个设计的关键在于： 所有条件判断都返回字符串，而非布尔值 。LangGraph 的 add_conditional_edges 要求 condition 函数返回字符串，该字符串必须是图中已定义的节点名或 END 。如果返回 True/False ，图会静默忽略，流程卡死在当前节点。我们在线上监控中专门添加了日志埋点，在 should_enable_hyde 中记录返回值，确保 100% 覆盖所有分支。

3.4 错误注入测试：用 httpx 模拟真实世界的服务故障

生产环境的稳定性，不取决于一切顺利时的表现，而取决于故障发生时的韧性。我们为系统编写了完整的错误注入测试套件，核心是 httpx 的 MockTransport ：

import httpx
from unittest.mock import Mock

# 模拟 OpenAI API 故障
def mock_openai_transport():
    transport = httpx.MockTransport(lambda request: httpx.Response(
        status_code=500,
        json={"error": {"message": "Internal Server Error"}}
    ))
    return transport

# 模拟 Chroma 向量库超时
def mock_chroma_transport():
    transport = httpx.MockTransport(lambda request: httpx.Response(
        status_code=408,
        text="Request Timeout"
    ))
    return transport

# 测试用例：验证系统在 OpenAI 故障时的降级能力
def test_openai_failure_fallback():
    # 创建一个降级的 LLM（使用本地 Phi-3）
    fallback_llm = ChatOllama(
        model="phi3:latest",
        temperature=0.1,
        num_predict=512
    )
    
    # 注入故障 transport
    client = openai.OpenAI(
        api_key="fake",
        base_url="https://api.openai.com/v1",
        http_client=httpx.Client(transport=mock_openai_transport())
    )
    
    # 执行工作流
    result = app.invoke({
        "input": "合同解除的条件是什么？",
        "hyde_enabled": False
    })
    
    # 断言：系统应返回降级结果，而非抛出异常
    assert result["analysis_result"] is not None
    assert "phi3" in result["analysis_result"].lower()  # 确认使用了降级模型

# 测试用例：验证向量库故障时的缓存兜底
def test_chroma_failure_cache():
    # 清空 Redis 缓存，确保无模板
    redis_client.flushdb()
    
    # 注入 Chroma 故障
    vectorstore = Chroma(
        client=chromadb.PersistentClient(
            path="/tmp/fake_path",  # 无效路径触发故障
            transport=mock_chroma_transport()
        ),
        collection_name="test",
        embedding_function=OpenAIEmbeddings()
    )
    
    # 执行工作流
    result = app.invoke({
        "input": "付款方式有哪些？",
        "hyde_enabled": False
    })
    
    # 断言：应返回错误消息，而非崩溃
    assert result.get("error") is not None
    assert "Chroma" in result["error"]

这些测试不是摆设。上线前，我们强制要求所有节点都通过 5 种故障模式的注入测试： API 超时、API 5xx 错误、向量库连接拒绝、Redis 缓存穿透、LLM 输出格式错误 。每个测试都验证两点：1）系统不崩溃，返回结构化错误；2）错误消息对用户友好（如“正在努力联系法律数据库，请稍后再试”而非“ConnectionRefusedError”）。正是这套测试，让我们在客户服务器突发网络抖动时，系统自动降级到本地 Phi-3 模型，保持了 99.2% 的请求成功率，避免了一次 P0 级事故。

4. 常见问题与排查技巧实录：那些文档里绝不会写的血泪教训

4.1 “No runnable found for tool name” —— Tool 注册的隐形陷阱

这是 LangChain Agent 新手最常遇到的报错，表面看是工具名不匹配，根源却在 Tool 类的 name 属性与 AgentExecutor 的 tools 列表注册方式不一致。我们曾在一个项目中， Tool 类定义为：

class SearchTool(BaseTool):
    name = "search"  # 注意：这里是小写
    description = "搜索法律数据库"
    ...

而 AgentExecutor 的初始化代码却是：

tools = [SearchTool()]
agent = create_openai_functions_agent(
    llm, tools, prompt
)
# 注意：create_openai_functions_agent 内部会调用 tools[0].name
# 但如果 tools 列表里混入了其他工具，顺序错乱会导致 name 匹配失败

问题在于： create_openai_functions_agent 会遍历 tools 列表，为每个 Tool 生成一个 FunctionTool ，其 function.name 字段直接取自 tool.name 。但如果 tools 列表中有两个 name 相同的工具（比如复制粘贴失误），或者 tool.name 包含非法字符（如空格、中文）， FunctionTool 的注册就会失败，但错误被静默吞掉，直到 Agent 运行时才抛出 No runnable found 。

独家排查技巧 ：在 AgentExecutor 初始化后，立即打印所有注册工具的 name ：

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 打印所有工具名
print("Registered tools:")
for t in agent_executor.tools:
    print(f"  - {t.name} (type: {type(t).__name__})")

更彻底的解决方案是： 永远使用 @tool 装饰器，并显式指定 name 参数 ：

@tool(name="legal_search")  # 显式命名，避免类名污染
def search_legal_database(query: str) -> str:
    """搜索法律数据库，输入为自然语言问题"""
    ...

@tool 装饰器会自动处理 name 的标准化（转为小写、去空格），并确保 args_schema 正确生成。我们团队已将此定为铁律：所有新工具必须用 @tool ，禁用 BaseTool 子类。

4.2 RetrievalQA 返回空结果 —— 向量库与检索器的双重校验法

RetrievalQA 返回空字符串，90% 的情况不是代码问题，而是数据或配置问题。我们建立了一套“双重校验法”：

第一步：绕过 QA 链，直接测试向量库检索

# 直接调用 retriever
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
docs = retriever.invoke("合同解除的条件")
print(f"Retrieved {len(docs)} documents")
for i, d in enumerate(docs):
    print(f"Doc {i}: {d.page_content[:100]}...")

如果这一步返回空，问题在向量库：检查 collection.count() 是否为 0，检查 embedding_function 是否与索引时一致（常见错误：索引用 text-embedding-ada-002 ，检索用 text-embedding-3-small ）。

第二步：检查 QA 链的 prompt 模板 RetrievalQA 的 prompt 模板中， context 变量必须与检索器返回的 Document 列表结构匹配。默认模板是：

Use the following pieces of context to answer the question at the end.
{context}
Question: {question}
Helpful Answer:

但如果 retriever 返回的 Document 对象的 page_content 是空字符串（比如 PDF 解析失败）， {context} 就是空的，LLM 只能看到 Question ，自然无法回答。解决方案是： 自定义 prompt，添加 context 非空校验 ：

from langchain.prompts import PromptTemplate

custom_prompt = PromptTemplate.from_template(
    """你是一个法律问答助手。请严格按以下规则回答：
    - 如果【上下文】为空，则回答"未找到相关信息，请提供更多细节"
    - 如果【上下文】不为空，则基于【上下文】回答问题，不要编造
    【上下文】
    {context}
    【问题】
    {question}
    【回答】"""
)

第三步：检查 LLM 的 stop sequence 某些 LLM（如 Llama-3）会在生成 "未找到相关信息" 后，继续输出无关内容。我们在 ChatOpenAI 初始化时，强制设置 stop=["\n\n", "Question:", "【问题】"] ，确保回答在合理位置截断。

4.3 LangGraph 状态丢失 —— StateGraph 的深拷贝陷阱

在 LangGraph 中， State 是一个 TypedDict ，但 Python 的 dict 是可变对象。如果在节点函数中直接修改 state 的嵌套字典，会导致状态污染：

# 错误示范：直接修改 state
def bad_node(state: ContractState) -> ContractState:
    state["retrieved_clauses"].append(new_doc)  # 修改原对象！
    return state

# 正确示范：返回新字典
def good_node(state: ContractState) -> ContractState: