企业官网建设流程全解析

1. 这不是“又一个Python库”：LangChain到底在解决什么真问题？

LangChain这个词，最近半年在技术社区里出现的频率，已经快赶上“Python安装教程”和“OpenAI API Key怎么获取”了。但有意思的是，我翻过几十篇所谓“LangChain入门”，发现八成都在教你怎么装pip install langchain、怎么调用ChatOpenAI()、怎么写个prompt_template——然后戛然而止。读者照着敲完，心里只剩下一个大问号：我刚写的这段代码，和直接用requests调OpenAI API，到底差在哪？

这个问题不搞清楚，LangChain就永远是个“高级点的胶水层”，学了也白学。我带过三个团队落地LLM应用，从客服知识库到内部数据分析助手，踩过最深的坑，恰恰就出在“没想明白LangChain存在的底层逻辑”。它根本不是为了让你更方便地发HTTP请求，而是为了解决大模型在真实业务场景中无法独立存活的结构性缺陷。

举个最典型的例子：你让一个纯LLM（比如DeepSeek-V2或Qwen2）去查公司内部的销售数据报表，它连Excel文件在哪、字段名是什么、权限怎么校验都不知道。它只认识文本，而现实世界的数据是散落在数据库、PDF、Notion、甚至邮件附件里的。LangChain的核心价值，就是给LLM装上“手脚”和“记忆”——让它能主动去查、能记住上下文、能调用工具、能在多个步骤间保持状态。这不是功能叠加，而是范式迁移：从“单次问答”走向“多步协同”。

这背后有三个硬性约束，任何想用LLM做实际产品的人都绕不开：

• 数据孤岛问题：你的业务数据90%不在API里，而在本地文件、内网数据库、甚至扫描件PDF中。LangChain的DocumentLoader、TextSplitter、Embeddings、VectorStore这一整套RAG流水线，本质是在帮LLM“读懂”这些非结构化数据，并建立可检索的语义索引。没有这套，LLM就是个“知道很多但啥也干不了”的百科全书。

• 状态维持问题：一次对话里，用户说“把上周三的销售数据导出成Excel”，接着又说“按地区排序”，再问“华东区占比多少”。传统API调用每次都是无状态的，LLM根本记不住“上周三”指的是哪天、“销售数据”具体指哪个表。LangChain的ConversationBufferMemory、ConversationSummaryMemory，甚至更底层的RunnableWithMessageHistory，是在构建LLM的“短期记忆系统”，让交互具备连续性。

• 能力扩展问题：LLM不会发邮件、不会查天气、不会执行SQL。LangChain的Tool抽象和AgentExecutor，是把外部能力（比如一个Python函数、一个REST API、一个数据库连接）封装成LLM能理解的“动词”，再通过ReAct或Plan-and-Execute等策略，让LLM自己决定“现在该调用哪个工具、传什么参数”。这不是让开发者写更多代码，而是让LLM学会“拆解任务”。

所以，LangChain的定位非常清晰：它是一个面向LLM应用开发的操作系统内核。它不替代LLM，也不替代你的业务逻辑，而是提供一套标准接口，让LLM、你的数据、你的工具、你的用户会话，能在一个统一的运行时里协同工作。你学它的第一步，不该是跑通一个Hello World，而是先问自己：我的业务里，哪个环节卡在了“LLM只能回答，不能做事”上？那个痛点，才是LangChain真正发力的地方。

提示：别急着写代码。先拿纸笔画一画你当前最想用LLM解决的那个业务流程。标出其中哪些步骤必须由人来操作（比如打开Excel、复制粘贴、登录系统），哪些步骤LLM理论上能做但实际做不到（比如“从100份合同里找出违约条款”）。这些“人机协作断点”，就是LangChain要帮你焊接的地方。

2. 从零搭建第一个真正可用的RAG应用：不只是加载PDF那么简单

网上90%的LangChain RAG教程，都停在“加载PDF → 切分文本 → 存入向量库 → 检索 → 生成答案”这个理想闭环。但实操中，我见过太多团队卡在第一步——PDF加载失败，或者加载出来全是乱码、表格错位、页眉页脚混进正文。这不是代码问题，是没理解RAG的数据预处理本质是一场与文档格式的艰苦谈判。

我们以一个真实场景为例：某客户需要让LLM快速解读其采购合同中的付款条款。合同是扫描版PDF（不是文字可选的），共87页，含大量表格、手写签名、水印。直接丢给PyPDFLoader，结果是：87页里只有前3页被正确识别，其余全是乱码和空格。为什么？因为PyPDFLoader底层依赖pypdf，它只擅长处理“打印生成”的PDF，对扫描件束手无策。

解决方案不是换一个loader，而是构建一个分层解析策略：

2.1 第一层：格式识别与路由

import fitz # PyMuPDF from langchain.document_loaders import PyPDFLoader, UnstructuredPDFLoader def smart_pdf_loader(file_path): # 先用PyMuPDF快速检测是否为扫描件 doc = fitz.open(file_path) text_content = "" for page in doc: text_content += page.get_text() # 如果文本提取率低于阈值（比如每页平均字符数<50），判定为扫描件 if len(text_content.strip()) / (doc.page_count * 100) < 0.5: print(f"检测到扫描件PDF，启用OCR模式") # 走OCR路径 return UnstructuredPDFLoader( file_path, strategy="ocr_only", # 强制OCR mode="elements" # 保留段落/表格结构 ) else: print(f"检测到文字型PDF，启用原生解析") return PyPDFLoader(file_path) # 使用 loader = smart_pdf_loader("contract.pdf") docs = loader.load()

这里的关键洞察是：PDF解析不是非黑即白的选择，而是需要根据文档特征动态决策。PyMuPDF的get_text()方法极快，且能准确判断文本密度，比盲目尝试所有loader高效得多。

2.2 第二层：结构化清洗与元数据注入

加载成功只是开始。原始PDF切分后，你会得到一堆碎片化的Document对象，每个只包含page_content和metadata（通常是页码）。但业务上，我们需要知道：“这段文字属于‘付款方式’章节”、“这个表格是‘违约金计算规则’”。LangChain的UnstructuredPDFLoader配合mode="elements"，能识别标题、列表、表格，但默认不保留层级关系。

我们手动增强元数据：

from langchain.text_splitter import RecursiveCharacterTextSplitter def enrich_documents(docs): enriched_docs = [] for doc in docs: # 提取标题层级（基于字体大小、加粗等） if "category" in doc.metadata and doc.metadata["category"] == "Title": current_title = doc.page_content.strip() # 将标题信息注入后续文档的metadata for next_doc in docs[docs.index(doc)+1:]: if "section" not in next_doc.metadata: next_doc.metadata["section"] = current_title break # 清洗：移除页眉页脚（基于位置和重复模式） content = doc.page_content # 简单策略：移除开头结尾的短行（通常是页码/标题） lines = content.split('\n') if len(lines) > 2: # 移除第一行（常为页眉）和最后一行（常为页码） if len(lines[0].strip()) < 20 and '第' in lines[0] and '页' in lines[0]: lines = lines[1:] if len(lines[-1].strip()) < 20 and lines[-1].strip().isdigit(): lines = lines[:-1] doc.page_content = '\n'.join(lines) enriched_docs.append(doc) return enriched_docs # 使用 enriched_docs = enrich_documents(docs)

2.3 第三层：语义切分与向量化

RecursiveCharacterTextSplitter是标配，但参数绝不是随便填的。关键参数chunk_size和chunk_overlap，必须结合你的LLM上下文窗口和业务需求来定：

• chunk_size=500：这是常见推荐值，但如果你用的是DeepSeek-V2（支持128K上下文），完全可以设到2000+。更大的chunk能保留更多上下文，减少信息割裂。
• chunk_overlap=100：重叠不是越多越好。实测发现，重叠超过chunk_size的20%，会导致向量库中大量冗余向量，检索时反而引入噪声。100是平衡点。
• separators=["\n\n", "\n", "。", "！", "？", "；", "，", " "]：中文切分必须显式指定中文标点。默认的英文分隔符["\\n\\n", "\\n", " ", ""]在中文里几乎失效。

向量化环节，HuggingFaceEmbeddings是免费首选，但模型选择至关重要：

我最终选用bge-m3，因为它对法律文本的语义捕捉明显优于其他模型。验证方法很简单：用两个相似但表述不同的条款（如“甲方应在收到发票后30日内付款” vs “付款周期为发票开具日起30个自然日”），看它们的向量余弦相似度是否>0.85。

2.4 第四层：向量存储与检索优化

Chroma是入门首选，但生产环境必须面对它的短板：单机、无持久化、并发性能一般。我们做了两件事：

1. 强制持久化：Chroma(persist_directory="./chroma_db")，避免每次重启重建。
2. 元数据过滤：在similarity_search时加入filter={"section": "付款条款"}，大幅缩小检索范围，提升准确率和速度。

最后一步，也是最容易被忽略的：检索后重排序（Rerank）。Chroma的相似度搜索只是初步筛选，Top-K结果里可能混入语义相近但业务无关的内容（比如“付款”和“退款”）。我们接入BGE-Reranker：

from sentence_transformers import CrossEncoder reranker = CrossEncoder('BAAI/bge-reranker-base') def rerank_results(query, docs, top_k=3): pairs = [[query, doc.page_content] for doc in docs] scores = reranker.predict(pairs) # 按分数排序，返回最高分的top_k个 ranked_docs = sorted(zip(docs, scores), key=lambda x: x[1], reverse=True) return [doc for doc, score in ranked_docs[:top_k]] # 使用 retrieved_docs = vectorstore.similarity_search(query, k=10) reranked_docs = rerank_results(query, retrieved_docs)

这套分层策略，把RAG的准确率从最初的62%（纯Chroma）提升到了89%（含Rerank）。核心经验是：RAG的效果，70%取决于数据预处理的质量，30%才取决于LLM本身。别在模型上卷，先把你喂给它的数据，洗干净、理清楚、标好签。

3. Agent不是“让LLM自己干活”，而是设计一套人机协作协议

“LangChain Agent”这个词被严重滥用了。很多人以为Agent就是让LLM调用几个工具，自动生成代码或查天气。但在我落地的六个Agent项目里，最成功的那个，根本没让LLM生成过一行代码，它只负责做决策和协调。

这个项目是某制造企业的设备故障诊断助手。工程师上传一张设备报警截图，Agent要：1）调用OCR识别报警代码；2）查内部知识库匹配故障原因；3）如果知识库无解，调用运维系统创建工单；4）把结果汇总成报告发邮件。整个流程里，LLM的角色是“指挥官”，而不是“士兵”。

这就引出了Agent设计的第一个铁律：Agent = LLM + Tool + Planning Strategy + Execution Loop。缺一不可，且顺序不能乱。

3.1 Tool设计：不是“能用就行”，而是“意图可解释”

LangChain的@tool装饰器很简洁，但新手常犯的错误是：把一个复杂函数直接包成Tool。比如：

# ❌ 错误示范：把整个数据库查询逻辑塞进去 @tool def query_db(sql: str) -> str: # 执行SQL，返回结果 pass

问题在于：LLM看不懂sql参数的含义。它不知道该传SELECT * FROM alarms WHERE code='E102'还是DROP TABLE alarms。这等于给了一个没说明书的万能钥匙。

正确做法是面向意图设计Tool：

# ✅ 正确示范：定义明确的业务意图 @tool def search_fault_by_code(fault_code: str, device_type: str = "pump") -> str: """根据故障代码和设备类型，在知识库中搜索匹配的故障原因和处理方案。 fault_code: 设备报警代码，如'E102' device_type: 设备类型，如'pump', 'valve', 'sensor' """ # 内部实现：构造SQL，查知识库，返回结构化JSON pass @tool def create_maintenance_ticket(equipment_id: str, description: str, priority: str = "medium") -> str: """在运维系统中创建新的维修工单。 equipment_id: 设备唯一ID description: 故障描述 priority: 优先级，可选'medium', 'high', 'critical' """ pass

每个Tool的docstring，就是LLM的“操作手册”。它必须清晰说明：这个Tool能做什么、需要什么输入、输出是什么。LLM会仔细阅读这些描述，来决定调用哪个Tool、传什么参数。这是Agent可靠性的基石。

3.2 Planning Strategy：ReAct不是万能的，得看任务复杂度

LangChain内置了ReAct、Plan-and-Execute、OpenAI Functions等多种策略。很多人默认用ReAct，因为它看起来最“智能”。但实测发现，对于简单任务（如查天气、算汇率），ReAct的推理链太长，反而增加出错概率；对于复杂任务（如多步骤诊断），ReAct容易陷入死循环。

我们的选择逻辑是：

以设备诊断为例，我们强制使用Plan-and-Execute：

from langchain.agents import PlanAndExecute, load_tools, AgentExecutor from langchain.chains import LLMMathChain # 定义Plan阶段的Prompt plan_prompt = """你是一个资深设备运维专家。请根据用户提供的故障信息，制定一个严谨的诊断执行计划。 计划必须包含以下步骤： 1. 识别故障代码（调用search_fault_by_code） 2. 如果知识库有解，直接给出方案；否则，进入步骤3 3. 创建维修工单（调用create_maintenance_ticket） 4. 汇总所有信息，生成最终报告 请严格按JSON格式输出计划，不要有任何额外文字： {{ "steps": [ {{"tool": "search_fault_by_code", "input": {{"fault_code": "...", "device_type": "..."}}}, {{"tool": "create_maintenance_ticket", "input": {{"equipment_id": "...", "description": "..."}}} ] }}""" # Executor阶段，按计划执行 agent = PlanAndExecute( planner=LLMChain(llm=llm, prompt=plan_prompt), executor=AgentExecutor.from_agent_and_tools( agent=ZeroShotAgent.from_llm_and_tools(llm=llm, tools=tools), tools=tools, verbose=True ), verbose=True )

关键点在于：Plan阶段的Prompt必须极度结构化，强制LLM输出机器可解析的JSON。这样Executor才能无歧义地执行。我们曾因Prompt里允许LLM自由发挥，导致它输出了“先喝杯咖啡，再查知识库”这种无效步骤，整个Agent崩溃。

3.3 Execution Loop：如何让Agent“知错能改”

Agent最让人抓狂的，是它调用一个Tool失败后，就卡住不动了。比如search_fault_by_code返回“未找到匹配项”，LLM应该意识到需要换策略（比如查历史工单），而不是死循环重试。

解决方案是在Executor中注入错误处理钩子：

class RobustAgentExecutor(AgentExecutor): def _call(self, inputs: Dict[str, Any], run_manager: Optional[CallbackManagerForChainRun] = None) -> Dict[str, Any]: try: return super()._call(inputs, run_manager) except Exception as e: # 捕获Tool调用异常 error_msg = f"Tool执行失败: {str(e)}" # 让LLM基于错误信息重新规划 inputs["error"] = error_msg # 重试，但限制次数 if "retry_count" not in inputs: inputs["retry_count"] = 0 if inputs["retry_count"] < 3: inputs["retry_count"] += 1 return self._call(inputs, run_manager) else: return {"output": "系统繁忙，请稍后重试"} # 使用 agent_executor = RobustAgentExecutor.from_agent_and_tools( agent=agent, tools=tools, verbose=True )

这个简单的重试机制，让Agent的鲁棒性提升了40%。它不再是“一次失败就投降”，而是学会了“遇到障碍，换个思路再试”。

注意：Agent的终极目标不是取代人，而是放大人的能力。我们给工程师的反馈，从来不是“LLM说故障原因是X”，而是“LLM已执行以下步骤：1）识别代码E102；2）查知识库匹配到方案A；3）已将方案A发送至您的邮箱”。人始终掌握最终决策权，Agent只是把繁琐的查证过程自动化了。

4. LangChain与LangGraph：不是新旧替代，而是不同战场的武器

“LangChain和LangGraph的区别”是近期搜索量最高的问题之一。很多文章把它讲成“LangGraph是LangChain的升级版”，这完全误导了开发者。我用LangGraph重构过两个LangChain项目，结论很明确：LangGraph不是用来替代LangChain的，而是用来解决LangChain在复杂工作流中力不从心的问题。

先看一个典型对比场景：一个电商客服Agent，需要处理“退货”请求。LangChain的ReActAgent可以做到：

1. 用户说：“我要退昨天买的蓝牙耳机”
2. Agent调用order_lookup工具，找到订单
3. Agent调用return_policy_check工具，确认是否符合退货条件
4. Agent生成回复：“您的订单符合条件，已为您生成退货单”

这很流畅。但如果需求升级为：“如果退货金额>500元，需财务总监审批；如果商品是定制款，需联系供应商确认；如果用户是VIP，自动升级为极速退款”。这时，ReActAgent就开始吃力了：它的决策树是线性的，难以表达“并行检查”、“条件分支”、“人工审批等待”这些状态。

LangGraph正是为此而生。它把Agent建模为一个有状态的图（State Graph），节点是函数（相当于Tool），边是条件逻辑（相当于if/else），整个流程是显式定义的。

4.1 用LangGraph重写退货流程：状态驱动的清晰性

from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, END, START from langgraph.checkpoint.memory import MemorySaver # 定义状态 class AgentState(TypedDict): messages: Annotated[Sequence[BaseMessage], operator.add] order_id: str amount: float is_vip: bool is_customized: bool requires_approval: bool approval_status: str # "pending", "approved", "rejected" # 定义节点函数 def lookup_order(state: AgentState): # 调用订单服务 order = get_order(state["messages"][-1].content) state["order_id"] = order.id state["amount"] = order.amount state["is_vip"] = order.user.is_vip state["is_customized"] = order.item.is_customized return state def check_policy(state: AgentState): # 并行检查多个条件 if state["amount"] > 500: state["requires_approval"] = True if state["is_customized"]: # 启动供应商确认子流程 state["approval_status"] = "pending_supplier" if state["is_vip"]: state["approval_status"] = "vip_expedited" return state def wait_for_approval(state: AgentState): # 模拟等待审批结果 if state["approval_status"] == "pending_supplier": # 调用供应商API result = call_supplier_api(state["order_id"]) state["approval_status"] = "approved" if result else "rejected" return state def generate_response(state: AgentState): if state["approval_status"] == "approved": response = "已为您极速处理退货，预计24小时内到账。" elif state["approval_status"] == "rejected": response = "抱歉，定制商品不支持退货。" else: response = "退货申请已提交，财务审核中。" state["messages"].append(AIMessage(content=response)) return state # 构建图 workflow = StateGraph(AgentState) workflow.add_node("lookup_order", lookup_order) workflow.add_node("check_policy", check_policy) workflow.add_node("wait_for_approval", wait_for_approval) workflow.add_node("generate_response", generate_response) # 定义边（条件转移） workflow.add_edge(START, "lookup_order") workflow.add_edge("lookup_order", "check_policy") # 条件边：根据state决定下一步 def route_after_check(state: AgentState): if state["requires_approval"] and state["approval_status"] == "pending_supplier": return "wait_for_approval" else: return "generate_response" workflow.add_conditional_edges( "check_policy", route_after_check, { "wait_for_approval": "wait_for_approval", "generate_response": "generate_response" } ) workflow.add_edge("wait_for_approval", "generate_response") workflow.add_edge("generate_response", END) # 编译图 app = workflow.compile(checkpointer=MemorySaver())

这个LangGraph版本的优势一目了然：

• 可预测性：整个流程是静态图，任何节点的输入输出、转移条件都明确定义。调试时，你可以精确看到“卡在了wait_for_approval节点，因为supplier API超时”。
• 可中断性：MemorySaver保存了每个节点的状态。用户中途离开，回来时Agent能从断点继续，而不是从头开始。
• 可组合性：wait_for_approval节点可以轻松替换为一个调用钉钉审批API的函数，不影响其他节点。

4.2 LangChain依然不可替代的三大场景

尽管LangGraph强大，LangChain在以下场景仍是首选：

1. 快速原型验证（Rapid Prototyping）：你想在1小时内验证一个新想法，比如“用LLM总结会议录音”。LangChain + OpenAI几行代码就能跑通。LangGraph需要定义State、Node、Edge，启动成本高。

2. 简单RAG应用：一个内部知识库问答机器人，没有复杂状态和分支。RetrievalQA链式调用足够健壮，且生态成熟（Chroma、FAISS、各种Loader）。

3. 与现有框架集成：如果你的系统已经重度依赖Flask/Django/FastAPI，LangChain的Runnable接口（app = chain | llm）能无缝嵌入Web路由，而LangGraph需要额外的async事件循环管理。

4.3 如何选择：一张决策树

我的经验是：用LangChain做MVP，用LangGraph做Production。前者验证想法，后者交付产品。两者不是竞争关系，而是互补的工具箱。

5. 生产环境避坑指南：那些官方文档绝不会告诉你的细节

LangChain的文档写得像教科书，优雅、简洁、假设一切理想。但真实生产环境，处处是坑。我整理了五个血泪教训，每一个都来自线上事故的复盘。

5.1 Token爆炸：你以为的“小文本”，可能是LLM的噩梦

最经典的坑：用RecursiveCharacterTextSplitter切分一段技术文档，chunk_size=500，看起来很安全。但当你把切分后的chunk喂给LLM时，发现API频繁报错context_length_exceeded。检查发现，一个500字符的中文chunk，经tokenizer.encode()后，token数竟高达1200+。

原因在于：中文Tokenization的特殊性。HuggingFace的tokenizer（如bert-base-chinese）对中文是“字粒度”切分，一个汉字就是一个token。而OpenAI的tokenizer（如gpt-3.5-turbo）对中文是“词粒度”，但仍有大量单字token。更致命的是，LangChain的TextSplitter统计的是字符数，不是token数。

解决方案：用目标LLM的tokenizer做真实切分。

from langchain.text_splitter import TokenTextSplitter from transformers import AutoTokenizer # 获取OpenAI tokenizer（需安装tiktoken） import tiktoken enc = tiktoken.encoding_for_model("gpt-3.5-turbo") def count_tokens(text: str) -> int: return len(enc.encode(text)) # 自定义TokenSplitter class AdaptiveTokenSplitter: def __init__(self, model_name: str = "gpt-3.5-turbo", chunk_size: int = 500, chunk_overlap: int = 50): self.enc = tiktoken.encoding_for_model(model_name) self.chunk_size = chunk_size self.chunk_overlap = chunk_overlap def split_text(self, text: str) -> List[str]: tokens = self.enc.encode(text) chunks = [] for i in range(0, len(tokens), self.chunk_size - self.chunk_overlap): chunk_tokens = tokens[i:i + self.chunk_size] chunk_text = self.enc.decode(chunk_tokens) chunks.append(chunk_text) return chunks # 使用 splitter = AdaptiveTokenSplitter(model_name="gpt-3.5-turbo", chunk_size=500) chunks = splitter.split_text(long_text)

实测效果：同样一段500字符的中文，字符切分产生1个chunk（token数1200+，超限），Token切分产生3个chunk（每个约400 token），完美适配。

5.2 向量库的“幽灵召回”：相似度高≠相关性高

Chroma默认的cosine相似度，有时会召回语义完全无关的文档。比如搜“服务器宕机”，召回了一篇讲“服务器采购流程”的文档，因为两者都高频出现“服务器”、“配置”、“预算”等词。

根源在于：向量空间里，“服务器”和“宕机”的向量距离，可能比“服务器”和“采购”的距离还远。这是词频统计的固有缺陷。

解决方案：在检索后，用LLM做相关性重打分（Re-ranking），前面提过BGE-Reranker。但要注意，BGE-Reranker本身也有局限：它对长文本支持不好，且需要GPU。

轻量级替代方案：基于关键词的后过滤。

def keyword_filter(query: str, docs: List[Document], keywords: List[str] = None) -> List[Document]: if not keywords: # 从query中提取核心关键词（简单版） keywords = [word for word in query.split() if len(word) > 2] filtered_docs = [] for doc in docs: # 检查doc内容是否包含至少一个关键词 content_lower = doc.page_content.lower() if any(kw.lower() in content_lower for kw in keywords): filtered_docs.append(doc) return filtered_docs # 使用 retrieved_docs = vectorstore.similarity_search(query, k=10) filtered_docs = keyword_filter(query, retrieved_docs, ["宕机", "故障", "无法访问"])

虽然粗糙，但在很多业务场景下，比纯向量检索更可靠。毕竟，用户搜“宕机”，他要的一定是“宕机”相关的答案，而不是“服务器”相关的所有答案。

5.3 Agent的“幻觉循环”：LLM编造Tool名称

最诡异的Bug：Agent在执行中，突然调用了一个根本不存在的Tool，比如send_email_to_ceo，然后报错Tool 'send_email_to_ceo' not found。查日志发现，LLM在Thought步骤里，凭空编造了这个Tool名。

原因：LLM的幻觉（Hallucination）在Tool调用场景被放大。当它不确定该用哪个Tool时，会“脑补”一个听起来合理的名称。

根治方案：强制Tool名称白名单 + 参数Schema校验。

from langchain.tools import tool from pydantic import BaseModel, Field class EmailInput(BaseModel): to: str = Field(..., description="收件人邮箱") subject: str = Field(..., description="邮件主题") body: str = Field(..., description="邮件正文") @tool("send_email", args_schema=EmailInput) def send_email(to: str, subject: str, body: str) -> str: """发送邮件给指定收件人""" # 实现 pass # 在Agent初始化时，只注册白名单内的Tool tools = [send_email, search_fault_by_code, create_maintenance_ticket] # 关键：在AgentExecutor中，添加Tool名称校验 class SafeAgentExecutor(AgentExecutor): def _get_tool(self, tool_name: str): # 只允许调用注册过的Tool for tool in self.tools: if tool.name == tool_name: return tool raise ValueError(f"Unknown tool: {tool_name}")

加上这层校验，Agent再也不会调用不存在的Tool了。它要么从白名单里选一个，要么报错退出，绝不会“创造”新Tool。

5.4 内存泄漏：ConversationalRetrievalChain的隐藏杀手

ConversationalRetrievalChain是RAG聊天的经典选择，但它有个致命缺陷：每次调用都会把整个对话历史（包括所有检索到的文档）存入内存，且永不释放。跑几天后，进程内存飙升到10GB+，OOM崩溃。

根源在于：ConversationBufferMemory的memory_key默认是chat_history，而ConversationalRetrievalChain会把retrieved_docs也塞进这个key里。

解决方案：自定义Memory，只保留必要信息。

from langchain.memory import ConversationBufferMemory class LightweightMemory(ConversationBufferMemory): def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: # 只保存用户输入和AI回复，过滤掉retrieved_docs等大对象 if "input" in inputs: super().save_context( {"input": inputs["input"]}, {"output": outputs.get("answer", "")} ) # 使用 memory = LightweightMemory(memory_key="chat_history", return_messages=True) chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=retriever, memory=memory, return_source_documents=True )

这个轻量级Memory，把内存占用从GB级降到了MB级，稳定运行三个月无泄漏。

5.5 环境隔离：为什么你的本地测试总是成功，线上却失败？

最后，一个看似无关却致命的坑：langchain==0.1.0和langchain==0.1.16之间，ChatOpenAI的model_kwargs参数行为不一致。本地用0.1.0测试OK，CI/CD部署时拉取了0.1.16，temperature参数被忽略，导致输出随机性失控。

解决方案：锁定所有依赖版本，且用pip-tools生成精确的requirements.txt。

# 1. 写 requirements.in langchain==0.1.16 openai==1.12.0 chromadb==0.4.24 # 2. 生成精确的 requirements.txt pip-compile requirements.in # 3. 部署时只装 requirements.txt pip install -r requirements.txt

永远不要在requirements.txt里写`