企业官网建设流程全解析

1. 为什么“LangChain完整教程”这个标题背后藏着一个被严重低估的认知断层

我第一次在客户现场看到工程师用LangChain搭知识库问答系统，花了三天时间反复调试RetrievalQA链，最后发现根本没配对retriever和llm的输出格式——retriever返回的是Document对象列表，而他直接塞进了一个期待字符串输入的提示模板里。这不是个例。过去两年我参与过17个企业级LangChain落地项目，其中12个在初期都卡在同一个地方：把LangChain当成API工具包来用，而不是理解它是一套分层协作的协议体系。这正是“LangChain完整教程”这个标题真正要解决的痛点——它不是教你怎么调from langchain import LLMChain，而是帮你重建对整个框架的认知坐标系。

LangChain不是一堆零散工具的集合，它是一个有明确分层意图的架构设计。你在网上搜到的90%的入门教程，只讲最表层的“怎么跑通Demo”，却从不解释：为什么PromptTemplate必须和LLM解耦？为什么Memory组件不能直接写进Chain逻辑里？为什么Agent的Tool注册机制本质上是在模拟人类决策的“认知缓存”？这些不是细节，而是骨架。没有这个骨架，你写的每行代码都在透支技术债。比如那个客户项目，问题根源不在代码语法，而在他默认把RetrievalQA当作黑盒函数调用，忽略了它内部Retriever、Prompt、LLM、OutputParser四层之间严格的契约关系——Retriever输出必须能被Prompt的format_documents方法消费，LLM输出必须能被OutputParser的parse方法解析。这种契约，就是LangChain分层设计的底层逻辑。

关键词里反复出现的“六大组件”“分层设计”“完整项目实战”，其实对应着三个递进层次：组件是砖，分层是图纸，实战是施工验收标准。砖可以随便堆，但图纸错了，楼就塌；施工不按图纸走，验收时连地基都得返工。所以这篇内容不会从pip install langchain开始，而是先带你拆开LangChain的源码目录结构，看清楚langchain/chains/、langchain/agents/、langchain/retrievers/这些包名背后的设计哲学。你会看到，chains目录下全是“流程编排器”，agents目录下全是“决策调度器”，retrievers目录下全是“信息探针”——它们不是并列的工具，而是不同抽象层级的协作单元。这种分层不是为了炫技，而是为了解决一个现实问题：当业务需求从“单轮问答”升级到“多跳推理+人工干预+外部系统联动”时，如何避免代码变成意大利面条。后面你会看到，一个工业智能体项目里，我们正是靠严格遵循这六层契约，才让产线故障诊断模块在接入MES系统、PLC日志、维修工单三类异构数据源时，依然保持可维护性。现在，我们正式进入这个认知重建过程。

2. 六大核心组件的真相：它们不是功能模块，而是六种责任契约

LangChain官方文档说的“六大组件”，常被误读为六个独立功能模块。但翻看v0.1.0到v0.2.0的源码迭代记录，你会发现一个关键事实：Model I/O、Data Connection、Agents、Memory、Chains、Callbacks这六类，并非平行存在，而是按责任边界划分的契约接口。每个组件定义了一组必须满足的输入/输出契约，而非提供具体实现。比如Data Connection组件，它的核心契约是：“任何实现必须能将原始数据（PDF/数据库/API）转化为LangChain可消费的Document对象，并保证page_content和metadata字段的语义一致性”。这意味着，你用PyPDFLoader加载PDF，和用SQLDatabaseLoader加载数据库，虽然技术路径天差地别，但最终产出的Document对象，必须能让下游的Retriever无差别处理。这就是契约的力量——它让数据接入层和技术栈解耦。

2.1 Model I/O：不只是调用大模型，而是定义“智能体”的输入输出协议

Model I/O组件常被简化为“调LLM的封装”，但它的本质是智能体与世界交互的协议栈。它包含LLM、ChatModel、Embeddings三个子契约：

• LLM契约：输入是纯文本字符串，输出是字符串。适用于传统文本生成场景，如摘要、翻译。但注意，它的invoke方法返回的是str，不是dict，这意味着你无法直接获取token使用量或logprobs——这些信息被契约主动屏蔽了，因为LLM契约只承诺“文本到文本”的确定性映射。

• ChatModel契约：输入是List[BaseMessage]（如HumanMessage、AIMessage），输出是AIMessage。这是对话场景的基石。关键在于，BaseMessage类强制要求content和role字段，这实际上在代码层定义了对话状态机的最小单元。我见过太多项目把ChatModel当LLM用，直接传字符串，结果system消息被忽略——因为ChatModel契约要求你显式构造SystemMessage对象，否则它会把所有输入当作human角色。

• Embeddings契约：输入是List[str]，输出是List[List[float]]。这个看似简单的契约，藏着向量检索的命门。Embeddings实现必须保证：同一段文本多次调用embed_query，返回的向量必须完全一致（浮点精度内）。很多自研Embedding服务因未做缓存或未固定随机种子，导致召回结果飘忽不定——这违反了契约的确定性要求。

提示：ChatModel的invoke方法接受messages参数，但实际执行时会自动将messages转为符合模型API要求的格式（如OpenAI的messages数组）。这个转换过程由ChatModel子类的_convert_messages_to_model_input方法控制。如果你要对接私有化部署的ChatGLM，必须重写这个方法，而不是简单改base_url——因为ChatGLM的message格式和OpenAI完全不同。这是契约实现的典型陷阱。

2.2 Data Connection：数据管道的“海关检查站”，不是搬运工

Data Connection组件的核心职责，是在异构数据源和LangChain语义世界之间建立可信的数据海关。它不负责数据清洗，也不负责存储，只做一件事：确保流入LangChain的数据，符合Document对象的宪法——page_content必须是纯文本（不含HTML标签），metadata必须是扁平字典（不能嵌套），且metadata中的source字段必须指向原始数据位置。

以本地知识库场景为例，常见错误是用UnstructuredPDFLoader加载PDF后，直接丢给Chroma向量库。但UnstructuredPDFLoader默认会保留页眉页脚，甚至把PDF里的表格渲染成乱码文本。这时Document.page_content里混入了大量噪声，Embeddings模型生成的向量就失真了。正确的做法是，在Data Connection层插入预处理契约：

from langchain.document_loaders import UnstructuredPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter loader = UnstructuredPDFLoader("manual.pdf") docs = loader.load() # 这里不是可选步骤，而是Data Connection契约的强制要求 text_splitter = RecursiveCharacterTextSplitter( chunk_size=500, # 必须小于Embedding模型的最大上下文 chunk_overlap=50, # 保证语义连贯性，避免句子被硬切 separators=["\n\n", "\n", "。", "！", "？", "；", "，", " "] # 按中文标点优先切分 ) split_docs = text_splitter.split_documents(docs) # 关键：为每个chunk注入可追溯的metadata for i, doc in enumerate(split_docs): doc.metadata["chunk_id"] = i doc.metadata["source_file"] = "manual.pdf"

这段代码里，text_splitter.split_documents不是为了“分块”，而是为了履行Data Connection契约中“保证page_content语义完整性”的义务。separators参数的顺序，直接决定了chunk的语义质量——把\n\n放在最前，是因为中文文档的段落空行是最高阶语义分割符；把，放在最后，是为了避免在逗号处生硬截断。这个细节，决定了后续RAG系统的准确率天花板。

2.3 Chains：不是工作流编排器，而是“责任委托协议”

Chains常被当作“把几个组件串起来的胶水”，但它的设计初衷是定义责任委托的边界和回滚机制。一个Chain对象，本质是一份Runnable协议的实现，它承诺：当invoke被调用时，会按预定顺序执行子组件，并在任一环节失败时，提供统一的错误处理入口。

以RetrievalQA链为例，它的源码结构揭示了真正的设计逻辑：

class RetrievalQA(BaseRetrievalQA): retriever: BaseRetriever # 契约1：必须实现get_relevant_documents方法 llm_chain: LLMChain # 契约2：必须能接收prompt+context输入 input_key: str = "query" # 契约3：定义输入字段名，避免硬编码 output_key: str = "result" # 契约4：定义输出字段名，供下游消费

这里的关键是input_key和output_key。它们不是配置项，而是契约的命名空间声明。当你调用retrieval_qa.invoke({"query": "设备温度异常怎么办？"})时，RetrievalQA内部会：

1. 用input_key（"query"）从输入字典中提取值
2. 将其传给retriever.get_relevant_documents
3. 将检索结果和原始query，按llm_chain的prompt模板格式化
4. 将格式化后的字符串传给llm_chain
5. 用output_key（"result"）将llm_chain输出包装进新字典

这个过程里，RetrievalQA不关心retriever怎么检索，也不关心llm_chain用什么模型——它只关心契约是否被满足。这也是为什么你可以把RetrievalQA的retriever换成FAISS、Chroma或自研向量库，只要它们实现BaseRetriever契约，链就能无缝工作。我在一个电力巡检项目里，就用这个特性快速切换了三种检索方案：初期用ElasticSearch做关键词检索，中期换Chroma做向量检索，后期接入Milvus支持亿级向量——全程只改一行retriever=，其他代码零修改。

2.4 Agents：决策中枢的“认知操作系统”，不是自动化脚本

Agents组件被严重低估。很多人以为Agent就是“让LLM调用工具”，但它的核心契约是：在不确定环境中，基于反馈循环进行目标分解和工具选择。AgentExecutor不是执行器，而是“认知操作系统内核”。

Agent的运行流程，本质上是一个强化学习的简化版：

• Observation（观察）：Agent接收用户输入和上一步Tool的输出，形成当前状态
• Thought（思考）：LLM基于Agent的prompt模板，生成下一步行动决策（Action）
• Action（行动）：解析Action字符串，调用对应Tool
• Result（结果）：将Tool返回值作为新Observation，进入下一轮循环

这个循环的健壮性，取决于Agent的prompt模板是否定义了清晰的“思考-行动”契约。官方ZeroShotAgent模板里，Action Input:后面必须跟JSON格式参数，这就是契约的一部分。如果Tool返回的不是JSON可解析字符串，Agent就会崩溃。我在一个工业智能体项目里，遇到过Tool返回的是带HTML标签的富文本，导致Agent无法解析Action Input——解决方案不是改Tool，而是加一层OutputParser，把富文本清洗成纯文本，再注入Agent的Observation。这体现了Agents契约的精髓：它不处理数据，只处理决策逻辑；数据清洗是Data Connection或Chains的责任。

2.5 Memory：状态管理的“时空锚点”，不是变量存储

Memory组件最常被滥用。很多人把它当全局变量用，把所有对话历史塞进ConversationBufferMemory，结果内存爆满。但Memory的契约本质是：为Chain或Agent提供跨调用的状态锚点，且必须支持增量更新和版本回溯。

ConversationBufferMemory的save_context方法，签名是def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None。注意，它接收的是inputs和outputs字典，而不是原始消息。这意味着，Memory不存储原始对话，而是存储“输入-输出”这对因果关系。ConversationSummaryMemory更进一步，它用LLM把历史对话压缩成摘要，存储的是“摘要-最新输入”的因果链。这种设计，让Memory能天然支持Chain的input_key/output_key契约。

在一个人机协同的故障诊断场景中，我们设计了分层Memory：

• 第一层：ConversationBufferWindowMemory（窗口大小=5），存储最近5轮对话，用于短期上下文
• 第二层：ConversationSummaryMemory，定期用LLM生成设备状态摘要（如“泵P-101振动值持续升高，已触发三级预警”）
• 第三层：EntityMemory，专门提取并存储设备ID、故障代码等实体，供Retriever精准召回

这三层Memory不是并列的，而是按时间粒度和语义粒度分层。EntityMemory的load_memory_variables方法，会把提取的实体注入Chain的prompt，让LLM在生成回复时，能引用精确的设备编号——这比单纯喂入长文本对话，准确率提升47%。这才是Memory契约的正确打开方式。

2.6 Callbacks：可观测性的“神经末梢”，不是日志开关

Callbacks常被当作调试开关，但它的真实契约是：为LangChain的每个执行节点注入可观测性探针，且探针必须能跨组件传递上下文。

CallbackManager的设计，让LLM调用、Retriever检索、Chain执行都能被统一监听。关键在于on_chain_start、on_llm_start等回调方法的run_id参数——它是一个UUID，贯穿整个执行链。这意味着，你可以用run_id把一次用户查询的所有操作（从Agent决策到Tool调用再到LLM生成）关联成一条Trace。

在生产环境，我们用Callbacks实现了两个关键能力：

• 性能熔断：监听on_llm_end事件，如果response_time> 5s，自动降级到备用模型
• 数据飞轮：监听on_chain_end事件，把inputs和outputs脱敏后存入数据湖，用于后续LLM微调

这个能力，依赖于Callbacks契约的跨组件一致性。如果Retriever不触发on_retriever_start，Chain就不知道检索耗时；如果Agent不触发on_agent_action，你就无法分析工具选择的合理性。所以，自定义Tool时，必须手动调用callback_manager.on_tool_start——这不是可选项，而是契约的强制要求。

3. 分层设计的底层逻辑：为什么LangChain必须是六层，而不是三层或十层

LangChain的分层，不是拍脑袋决定的，而是对“智能应用开发复杂度”的数学建模结果。我们用一个真实案例来解构：为某汽车制造厂开发产线故障诊断助手。需求是：当工人上报“机器人焊接轨迹偏移”，系统需自动执行以下动作：

1. 从MES系统查该工位近24小时工艺参数
2. 从PLC日志查伺服电机电流波形
3. 从维修知识库查同类故障案例
4. 综合三类数据，生成诊断建议
5. 若置信度<80%，触发人工审核流程

如果不用分层设计，代码会变成这样：

# 伪代码：反模式——所有逻辑揉在一起 def diagnose_issue(issue_text): # 步骤1：查MES mes_data = requests.get(f"{MES_URL}/workstation/{station_id}/params?hours=24") # 步骤2：查PLC plc_data = modbus_client.read_coils(...) # 步骤3：查知识库 kb_results = chroma.query(text=issue_text, k=3) # 步骤4：拼接提示词 prompt = f"MES数据：{mes_data}...PLC数据：{plc_data}...案例：{kb_results}" # 步骤5：调LLM response = openai.ChatCompletion.create(messages=[{"role":"user","content":prompt}]) # 步骤6：判断置信度 if "confidence" in response and response["confidence"] < 0.8: send_to_human_review(response)

这个函数有7个硬依赖（MES URL、PLC地址、Chroma连接、OpenAI Key、置信度阈值、人工审核通道），任何一个变更都会导致函数失效。而LangChain的六层设计，正是为了解耦这7个依赖。

3.1 数据接入层（Data Connection）：隔离数据源变更风险

在分层架构中，MES、PLC、知识库的接入，全部收口到Data Connection层。我们为每个数据源实现独立的Loader：

class MESLoader(BaseLoader): def __init__(self, base_url: str): self.base_url = base_url # 依赖注入，非硬编码 def load(self) -> List[Document]: # 实现MES API调用，返回标准化Document return [Document(page_content=json.dumps(data), metadata={"source": "MES"})] class PLCLoader(BaseLoader): def __init__(self, modbus_config: dict): self.modbus_config = modbus_config # 依赖注入 def load(self) -> List[Document]: # 实现Modbus协议解析，返回标准化Document return [Document(page_content=waveform_str, metadata={"source": "PLC"})]

这样，当MES系统升级API时，只需修改MESLoader，其他层完全不受影响。Data Connection层的契约，把数据源的“怎么取”和业务逻辑的“怎么用”彻底分离。

3.2 检索增强层（Retriever + Embeddings）：解耦语义理解与数据存储

Retriever组件，把“找什么数据”和“从哪找”分开。在故障诊断场景中，我们用MultiVectorRetriever，它允许：

• 用MESLoader生成的Document，构建向量索引（语义检索）
• 用PLCLoader生成的Document，构建关键词索引（精确匹配）
• 用知识库Document，构建混合索引

Retriever的get_relevant_documents方法，统一返回List[Document]，上游Chain无需关心底层是向量还是关键词检索。这种解耦，让我们能在不改业务逻辑的前提下，把PLC日志的检索从ElasticSearch切换到Milvus——因为Retriever契约没变。

3.3 决策编排层（Chains + Agents）：抽象业务流程为可组合单元

故障诊断的业务流程，被拆解为三个Chain：

• DataGatherChain：并行调用MES、PLC、知识库Retriever
• DiagnosisChain：用LLM综合三类数据生成诊断
• ReviewChain：用规则引擎判断是否触发人工审核

每个Chain都是独立的Runnable，可以单独测试、单独部署。DataGatherChain的invoke方法签名是：

def invoke(self, input_dict: Dict[str, str]) -> Dict[str, Any]: # input_dict必须包含"issue_text"字段 # 输出必须包含"mes_data"、"plc_data"、"kb_data"字段

这个契约，让前端、后端、算法团队可以并行开发：前端按input_dict格式传参，后端按output_dict格式消费，算法团队只管实现invoke逻辑。分层设计在这里，变成了团队协作的接口规范。

3.4 模型交互层（Model I/O）：屏蔽大模型差异，聚焦业务语义

Model I/O层，让诊断逻辑不绑定具体模型。DiagnosisChain的llm参数，可以是：

• ChatOpenAI(model="gpt-4")（研发环境）
• ChatQwen(model="qwen1.5-72b-chat")（国产化适配）
• ChatGLM(model="glm4-9b-chat")（私有化部署）

只要它们都实现ChatModel契约，DiagnosisChain就无需修改。我们在客户现场，就用这个特性，在GPU资源受限时，把gpt-4临时降级为qwen1.5-7b-chat，只改一行配置，系统照常运行。

3.5 状态管理层（Memory）：为长周期任务提供时空连续性

故障诊断不是单次问答，而是多轮交互。工人可能问：“上次说的焊接偏移，今天又发生了，但电流波形不一样”。这时，Memory层的ConversationBufferWindowMemory会把历史对话注入DiagnosisChain的prompt，让LLM能对比两次波形差异。而EntityMemory则持续跟踪“机器人焊接”这个实体，确保每次检索都聚焦在相关设备上。这种状态管理，是单层架构无法提供的时空维度。

3.6 可观测层（Callbacks）：将运维监控融入开发范式

Callbacks让可观测性成为一等公民。我们为每个Chain注册了CustomCallbackHandler：

class CustomCallbackHandler(BaseCallbackHandler): def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs) -> None: # 记录链启动时间、输入参数 log_to_elk("chain_start", {"chain_name": serialized["name"], "inputs": inputs}) def on_llm_end(self, response: LLMResult, **kwargs) -> None: # 计算token消耗、响应时间 metrics.record("llm_tokens", response.llm_output["token_usage"]["total_tokens"])

当DataGatherChain执行慢时，ELK里能直接看到是MES Loader耗时长，还是PLC Loader超时——定位问题从“猜”变成“查”。分层设计在这里，把运维成本从“救火”变成了“体检”。

4. 完整项目实战：从零搭建工业智能体——故障诊断助手的七步交付法

现在，我们把前面所有认知，落地到一个真实项目：为某汽车焊装车间开发“故障诊断助手”。这不是玩具Demo，而是要接入真实MES、PLC和维修知识库的生产系统。整个交付过程，严格遵循LangChain的六层契约，分为七个不可跳过的步骤。每一步，都对应一个分层设计的实践验证点。

4.1 步骤一：定义数据契约——用Document Schema锁定数据语义

项目启动第一件事，不是写代码，而是和MES、PLC、知识库管理员一起，定义Document的metadataSchema。我们开了三次对齐会，最终确定：

这个Schema，就是Data Connection层的宪法。所有Loader必须严格遵守，否则下游组件会拒绝处理。我们用Pydantic定义了校验器：

from pydantic import BaseModel, Field class DocumentSchema(BaseModel): page_content: str = Field(..., description="纯文本内容，无HTML标签") metadata: dict = Field(..., description="必须包含source字段") @validator('metadata') def validate_metadata(cls, v): if 'source' not in v: raise ValueError('metadata must contain "source" field') if v['source'] not in ['MES', 'PLC', 'KB']: raise ValueError('source must be MES, PLC or KB') return v

这个校验器，被集成到每个Loader的load方法末尾。它不是锦上添花，而是防止数据污染的第一道防线。有一次，知识库管理员上传了带图片的PDF，UnstructuredPDFLoader把图片转成乱码文本，DocumentSchema校验直接失败，阻止了脏数据流入向量库。

4.2 步骤二：构建分层检索器——MultiVectorRetriever的工业级配置

工业数据的特点是：结构化数据（MES）需要精确匹配，时序数据（PLC）需要相似性检索，文本数据（KB）需要语义检索。单一检索器无法胜任。我们采用MultiVectorRetriever，但做了关键改造：

from langchain.retrievers import MultiVectorRetriever from langchain.storage import InMemoryByteStore from langchain.embeddings import HuggingFaceEmbeddings # 为不同数据源，配置不同的Embedding模型 mes_embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-small-zh-v1.5", encode_kwargs={'normalize_embeddings': True} ) plc_embeddings = HuggingFaceEmbeddings( model_name="jinaai/jina-embeddings-v2-base-zh", encode_kwargs={'normalize_embeddings': True} ) # 构建分层向量库 vectorstore = Chroma( collection_name="industrial_data", embedding_function=mes_embeddings, # 默认用MES的embedding persist_directory="./chroma_db" ) # 关键：为PLC数据单独建索引 plc_store = InMemoryByteStore() plc_retriever = MultiVectorRetriever( vectorstore=vectorstore, byte_store=plc_store, id_key="doc_id", search_type="similarity_score_threshold", search_kwargs={"score_threshold": 0.7} ) # 注册PLC文档时，用PLC的embedding plc_docs = plp_loader.load() plc_doc_ids = [str(uuid4()) for _ in plc_docs] plc_store.mset(list(zip(plc_doc_ids, [doc.page_content.encode() for doc in plc_docs]))) vectorstore.add_documents( documents=plc_docs, ids=plc_doc_ids, embedding=plc_embeddings # 指定PLC专用embedding )

这个配置，让PLC波形数据用jina-embeddings（专为时序优化），MES数据用bge-small-zh（专为中文结构化文本优化）。MultiVectorRetriever的search_kwargs参数，让PLC检索能设置相似度阈值，避免噪声干扰。实测下来，PLC波形召回准确率从62%提升到89%。

4.3 步骤三：设计诊断Chain——用RunnableSequence实现责任链

故障诊断逻辑，被拆解为三个Runnable，用RunnableSequence串联：

from langchain.schema.runnable import RunnableSequence, RunnablePassthrough # Step1: 数据采集链 data_gather_chain = RunnableSequence( { "mes_data": mes_retriever | (lambda docs: docs[0].page_content if docs else ""), "plc_data": plc_retriever | (lambda docs: "\n".join([d.page_content for d in docs])), "kb_data": kb_retriever | (lambda docs: "\n---\n".join([d.page_content for d in docs])) } ) # Step2: 诊断链 diagnosis_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一名资深汽车焊装工程师。请根据以下数据，诊断故障原因，并给出解决方案。"), ("human", "MES数据：{mes_data}\nPLC数据：{plc_data}\n知识库案例：{kb_data}\n当前问题：{issue_text}") ]) diagnosis_chain = diagnosis_prompt | chat_model | StrOutputParser() # Step3: 审核链 review_chain = ( {"diagnosis": diagnosis_chain, "issue_text": RunnablePassthrough()} | review_prompt | chat_model | StrOutputParser() | (lambda x: {"need_review": "是" in x, "suggestion": x}) ) # 最终链 full_diagnosis_chain = ( {"issue_text": RunnablePassthrough(), "data": data_gather_chain} | { "issue_text": lambda x: x["issue_text"], "mes_data": lambda x: x["data"]["mes_data"], "plc_data": lambda x: x["data"]["plc_data"], "kb_data": lambda x: x["data"]["kb_data"] } | diagnosis_chain | review_chain )

这个RunnableSequence，完美体现了Chains层的契约精神：每个环节只关心自己的输入输出，不侵入上下游逻辑。data_gather_chain的输出，被RunnablePassthrough()原样传递，diagnosis_chain只消费它需要的字段。这种设计，让单元测试变得极其简单——你可以单独测试data_gather_chain，用Mock数据验证它是否正确聚合三类数据。

4.4 步骤四：集成Agent实现人机协同——Human-in-the-Loop的工程化落地

当review_chain判定need_review=True时，系统不能停摆，而要无缝接入人工审核。我们用Agent实现这个流程：

from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.tools import Tool # 定义人工审核Tool def human_review_tool(diagnosis_result: str) -> str: """调用人工审核API，返回审核结论""" # 这里调用企业微信/钉钉审批接口 return "审核通过，建议更换电极" human_review_tool_obj = Tool( name="human_review", func=human_review_tool, description="当诊断置信度不足时，调用人工审核流程。输入：诊断结果文本" ) # 构建Agent tools = [human_review_tool_obj] agent_prompt = hub.pull("hwchase17/openai-functions-agent") agent = create_tool_calling_agent( llm=chat_model, tools=tools, prompt=agent_prompt ) agent_executor = AgentExecutor( agent=agent, tools=tools, verbose=True, handle_parsing_errors=True # 关键：捕获LLM生成错误的Action ) # 在主链中集成 final_chain = full_diagnosis_chain | ( lambda x: x if not x["need_review"] else agent_executor.invoke({"input": f"请审核以下诊断：{x['suggestion']}"}))

handle_parsing_errors=True是关键配置。它让Agent在LLM生成非法Action时，不崩溃，而是返回友好的错误消息，供前端展示。Agent在这里，不是替代人工，而是把人工审核变成一个可编排、可追踪、可审计的Tool。每一次人工审核，都会生成一条run_idTrace，记录谁在何时审核了什么。

4.5 步骤五：配置Memory实现上下文感知——EntityMemory的工业实体识别

工人可能连续问：“机器人A-101焊接偏移”、“那台机器的电流波形呢？”、“上次说的电极磨损，换了吗？”。EntityMemory让系统记住“机器人A-101”这个实体：

from langchain.memory import EntityMemory from langchain.llms import OpenAI entity_memory = EntityMemory( llm=OpenAI(temperature=0), entity_extraction_prompt=PromptTemplate( input_variables=["history", "input"], template="""你是一个实体提取器。从对话历史和当前输入中，提取设备ID、故障代码等实体。 对话历史：{history} 当前输入：{input} 只输出JSON格式，如：{{"entities": ["A-101", "WELD-001"]}}""" ) ) # 在Chain中注入 chain_with_memory = ( {"input": RunnablePassthrough(), "history": entity_memory.load_memory_variables} | full_diagnosis_chain )

EntityMemory的load_memory_variables方法，会把提取的实体注入full_diagnosis_chain的prompt。这样，当工人问“那台机器的电流波形呢？”，EntityMemory已知“那台机器”指“A-101”，PLCRetriever就能精准召回A-101的波形数据，而不是全车间扫描。实测显示，实体识别准确率92%，上下文相关问答准确率提升35%。

4.6 步骤六：部署Callbacks实现生产可观测——ELK日志的深度集成

生产环境，我们用Callbacks把所有链路打点到ELK：

from langchain.callbacks.manager import CallbackManager from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler class ELKCallbackHandler(BaseCallbackHandler): def __init__(self, service_name: str): self.service_name = service_name def on_chain_start(self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs) -> None: