LangChain在数据工程中的生产级落地：从Prompt管理到可观测性

1. 这不是又一个“AI框架”介绍——它是我亲手用LangChain重构三条数据流水线后，才敢写的实操手记

你肯定见过太多标题党：“5分钟上手LangChain”、“LangChain保姆级教程”、“一文看懂LangChain核心原理”。我过去三年在金融、电商、SaaS三类企业做过数据平台架构，也带过十几支数据工程团队。坦白说，90%的LangChain文章，写完就扔进回收站——它们连最基础的“为什么非要用LangChain而不是直接调OpenAI API”都讲不清，更别说告诉你：什么时候该用RetrievalQAChain，什么时候必须自己写LLMChain+自定义OutputParser；为什么在ETL任务里硬套ConversationalRetrievalChain反而会让pipeline卡死；以及——最关键的一点——LangChain根本不是为“写个聊天机器人”设计的，它是为“把大模型塞进生产级数据系统”而生的工业级胶水层。

这篇文章，就是我用LangChain重写了三套真实业务系统后的复盘：一套是某银行信用卡中心的实时风控规则解释引擎（日均处理27万条交易），一套是某跨境电商的多源商品知识图谱构建流水线（接入14个API+8类结构化/非结构化数据源），还有一套是内部BI平台的自然语言查询服务（支持SQL生成+结果解读+可视化建议）。全文不讲概念、不画架构图、不堆API列表。只讲三件事：第一，它到底解决了数据工程师每天被卡住的哪几个具体痛点；第二，每个痛点背后的真实代码片段、参数选择依据和性能实测数据；第三，我踩过的坑、改过的源码、以及现在每次新建项目必加的5行防御性配置。

如果你正在评估是否在下一个数据项目中引入LangChain，或者已经写了几十行llm.predict()却卡在“怎么让输出稳定成JSON”、“怎么把数据库表结构喂给模型还不超token”、“怎么让模型记住上一轮分析结论”，那这篇就是为你写的。它不教你怎么安装pip install langchain，但会告诉你：为什么langchain==0.1.16比0.1.20更适合你的Spark作业；为什么Chroma向量库在千万级文档下必须关掉persist_directory；以及——当你发现ConversationBufferMemory在高并发下内存暴涨时，真正该改的不是max_token_limit，而是input_key和output_key的字段映射逻辑。

关键词：LangChain、数据工程、LLM应用、数据管道、提示工程、输出解析、向量索引、链式调用、Agent调度、生产部署。

2. LangChain解决的从来不是“调用大模型”，而是“让大模型在数据系统里活下来”

2.1 痛点一：Prompt里的“废话文学”正在吃掉你80%的开发时间

别笑。我刚接手那个银行风控项目时，原始需求很简单：“当交易触发规则X时，用自然语言生成一段向客户解释原因的话”。开发同学第一天交来的代码是这样的：

def generate_explanation(transaction_id: str, rule_name: str) -> str: prompt = f""" 你是一名资深银行风控专家，语气专业但亲切，避免使用术语。 请用不超过120字向客户解释：为什么交易{transaction_id}被规则{rule_name}拦截？ 要求：1. 必须基于事实；2. 不得承诺任何补偿；3. 引导客户联系客服。 """ response = openai.ChatCompletion.create( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content.strip()

看起来很干净？问题出在第二周：业务方要求增加“多语言支持”，第三周要求“不同客群语气差异化”（VIP客户更简洁，新用户更详细），第四周要求“嵌入实时账户余额信息”。于是prompt从12行膨胀到83行，每次修改都要全量回归测试——因为哪怕少一个句号，GPT-4的输出格式就可能从“您的交易因……”变成“【风险提示】您的交易……”。

LangChain的PromptTemplate不是语法糖，它是把Prompt当作可版本管理、可单元测试、可A/B分流的配置项。我们最终落地的方案是：

from langchain.prompts import PromptTemplate from langchain.chains import LLMChain # 把所有“废话”抽成模板变量 explanation_template = PromptTemplate( input_variables=["rule_name", "transaction_id", "customer_tier", "balance"], template="""你是一名{customer_tier}级别的银行风控专家，语气{tone}。 当前交易{transaction_id}被规则{rule_name}拦截。 账户余额：{balance}元。 请用{word_count}字内向客户解释原因，要求： 1. 基于事实，不虚构规则细节； 2. 不承诺补偿或特殊处理； 3. 结尾引导联系955XX客服。 """ ) # tone和word_count由客户等级动态决定 tier_config = { "vip": {"tone": "简洁有力", "word_count": 80}, "new": {"tone": "耐心细致", "word_count": 120}, "standard": {"tone": "专业中立", "word_count": 100} } # 实际调用时只传业务参数 chain = LLMChain(llm=llm, prompt=explanation_template) result = chain.run( rule_name="high_freq_small_amt", transaction_id="TXN-88234", customer_tier="vip", balance=12500.50, **tier_config["vip"] # 动态注入tone和word_count )

提示：PromptTemplate真正的威力在于和FewShotPromptTemplate组合。我们在风控场景中预置了200+条人工审核过的“优质解释样本”，用ExampleSelector按规则类型自动匹配最相似的3个例子，再注入prompt。实测将GPT-4的解释准确率从72%提升到91%，且完全规避了“编造规则细节”的致命错误。

2.2 痛点二：模型返回的“文本”不是终点，而是你下游系统的噩梦起点

数据工程师最怕什么？不是SQL写错，而是上游系统返回的“看似正确”的字符串。比如这个需求：“从客服对话日志中提取客户投诉的3个核心问题，以JSON格式返回”。

直接调API的代码：

response = llm.invoke("请提取以下对话中的3个核心问题，返回JSON格式：{dialogue}") # response可能是： # {"issues": ["账单错误", "退款延迟", "客服态度差"]} # 也可能是： # 问题1：账单错误\n问题2：退款延迟\n问题3：客服态度差 # 甚至： # 我理解您的三个主要问题：1）账单错误；2）退款延迟；3）客服态度差。

LangChain的OutputParser不是做字符串切割，而是构建从非结构化文本到强类型对象的可信转换通道。我们选型时对比了三种方案：

最终采用的PydanticOutputParser代码：

from pydantic import BaseModel, Field from langchain.output_parsers import PydanticOutputParser class ComplaintIssues(BaseModel): issues: list[str] = Field(description="客户投诉的3个核心问题，每个问题不超过10个字") confidence: float = Field(description="对提取结果的信心评分，0.0-1.0") parser = PydanticOutputParser(pydantic_object=ComplaintIssues) # 关键：用parse_with_prompt强制模型理解输出约束 prompt = PromptTemplate( template="请严格按JSON格式提取以下对话的核心问题：\n{dialogue}\n{format_instructions}", input_variables=["dialogue"], partial_variables={"format_instructions": parser.get_format_instructions()} ) chain = LLMChain(llm=llm, prompt=prompt) | parser result = chain.invoke({"dialogue": dialogue_text}) # result是ComplaintIssues实例，可直接存入数据库或传给下游

注意：get_format_instructions()生成的提示词不是固定字符串，它会根据Pydantic模型的Field描述动态生成。我们曾因忘记加description导致模型返回{"issues": []}空数组——因为模型根本不知道“issues”该填什么。这是新手最容易栽的坑。

2.3 痛点三：你以为在换模型，其实是在重写整个数据管道

很多团队说“我们先用GPT-4验证效果，再切到开源模型降本”。听起来很合理？现实是：当你把openai.ChatCompletion.create()换成llama_cpp.Llama()时，要改的不只是API地址。GPT-4的temperature=0.3在Llama-3-70B上可能输出完全随机，max_tokens=512在本地模型上可能直接OOM，更别说system角色消息在不同模型间的支持度差异。

LangChain的LLM抽象层价值在此刻爆发。我们为银行项目维护的模型工厂类：

from langchain.llms import OpenAI, LlamaCpp, HuggingFacePipeline from langchain_community.llms import Ollama class LLMFactory: @staticmethod def get_llm(model_type: str, **kwargs) -> BaseLLM: if model_type == "gpt-4": return OpenAI( model_name="gpt-4", temperature=0.2, # GPT-4需更低温度保事实性 max_tokens=1024, request_timeout=30 ) elif model_type == "llama3-70b": return LlamaCpp( model_path="/models/llama3-70b.Q4_K_M.gguf", n_ctx=4096, # 必须显式设置，否则默认2048 n_threads=16, temperature=0.1, # 开源模型需更低温控 max_tokens=512, verbose=False ) elif model_type == "phi-3-mini": return Ollama( model="phi3:3.8b", temperature=0.05, num_predict=256, format="json" # Phi-3原生支持JSON模式 ) else: raise ValueError(f"Unsupported model: {model_type}") # 在pipeline中统一调用 llm = LLMFactory.get_llm("llama3-70b") chain = LLMChain(llm=llm, prompt=prompt)

实操心得：不要迷信llm.predict()。在数据管道中，我们90%的场景用llm.generate()——因为它返回LLMResult对象，包含所有生成的候选序列（generations）、token统计（llm_output）和调试信息。当某次批量处理出现异常时，能立刻定位是模型崩溃、还是prompt超长、或是网络抖动。

3. 数据工程视角下的LangChain核心组件：不是功能罗列，而是故障树分析

3.1 向量索引（Index）：别再用“prompt stuffing”喂模型了

“把整张表塞进prompt”是数据工程师最常犯的错误。某电商客户曾想用GPT-4分析10万行商品评论，直接df.to_string()后拼进prompt——结果87%的请求因token超限失败，剩下13%的响应中，72%把“iPhone 15”识别成了“iPhone 14 Pro Max”。

LangChain的DocumentLoader+TextSplitter+VectorStore不是炫技，而是把数据从“不可计算”变成“可检索”的必要工序。我们为商品知识图谱项目设计的索引流程：

from langchain_community.document_loaders import DataFrameLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings # 1. 加载：支持DataFrame、CSV、JSON、PDF等12种格式 loader = DataFrameLoader( df=product_reviews_df, page_content_column="review_text" # 指定文本列 ) documents = loader.load() # 2. 切分：关键参数选择依据 text_splitter = RecursiveCharacterTextSplitter( chunk_size=512, # 为什么是512？因为我们的embedding模型（all-MiniLM-L6-v2）最大输入512token chunk_overlap=64, # 重叠64字符，避免语义断裂（如“价格$999”被切成“价格”和“$999”） separators=["\n\n", "\n", "。", "！", "？", "；", "，", " "] # 中文优先按标点切 ) texts = text_splitter.split_documents(documents) # 3. 向量化：生产环境必须关掉persist_directory！ embeddings = HuggingFaceEmbeddings( model_name="sentence-transformers/all-MiniLM-L6-v2" ) vectorstore = Chroma.from_documents( documents=texts, embedding=embeddings, # persist_directory="./chroma_db", # ❌ 高并发下文件锁导致阻塞 collection_name="product_reviews" ) # 4. 检索：不是简单search，而是带业务规则的rerank retriever = vectorstore.as_retriever( search_type="mmr", # 最大边缘相关性，避免重复主题 search_kwargs={"k": 5, "fetch_k": 20} # 先取20个，MMR重排选5个 )

关键经验：RecursiveCharacterTextSplitter的chunk_size必须与embedding模型的max_input_length严格对齐。我们曾用bge-large-zh（max 512）但设chunk_size=1024，结果embedding向量全是NaN。查了3天才发现是HuggingFace源码里有个未文档化的truncate=True默认开关。

3.2 链（Chain）与智能体（Agent）：何时该用线性链，何时必须上Agent？

很多教程把SequentialChain吹成银弹。但在真实数据管道中，90%的场景只需要LLMChain，5%需要TransformChain，剩下5%才轮到Agent。我们总结了一个决策树：

graph TD A[需求：用LLM处理数据] --> B{是否需要条件分支？} B -->|否| C[用LLMChain或SimpleSequentialChain] B -->|是| D{分支逻辑是否依赖LLM输出？} D -->|否| E[用Python if/else + 多个LLMChain] D -->|是| F[必须用Agent] F --> G{是否需调用外部工具？} G -->|否| H[ZeroShotAgent] G -->|是| I[ToolCallingAgent]

注：此处禁用Mermaid，改用文字描述

• LLMChain：单步调用，如“把SQL转成自然语言解释”
• SimpleSequentialChain：多步线性，如“先用LLM生成SQL → 再用pandas执行 → 最后用LLM解释结果”，但每步输入输出固定
• Agent：当LLM的输出决定下一步动作时启用，如“分析用户问题→判断需查数据库还是调API→执行对应操作→整合结果”。我们电商项目中，Agent负责动态选择：查MySQL商品表、调用ERP库存API、还是读取S3上的促销规则JSON。

实际Agent代码（简化版）：

from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.tools import Tool from langchain_core.prompts import ChatPromptTemplate # 定义工具 tools = [ Tool( name="sql_search", func=lambda q: run_sql_query(q), # 封装SQL执行 description="用于查询商品基本信息、库存、价格" ), Tool( name="api_search", func=lambda p: call_erp_api(p), # 封装ERP调用 description="用于获取实时库存和物流状态" ) ] # Agent提示词必须包含工具描述和调用格式 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个电商数据助手。只能使用以下工具：{tool_names}"), ("human", "{input}"), ("placeholder", "{agent_scratchpad}") # Agent自动填充的思考过程 ]) # 创建Agent agent = create_tool_calling_agent(llm, tools, prompt) agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 调用 result = agent_executor.invoke({ "input": "iPhone 15 Pro的当前库存和预计到货时间？" }) # 输出自动包含：思考过程 → 调用sql_search → 调用api_search → 整合回答

注意：Agent的verbose=True在开发期是救命稻草，但在生产环境必须关掉——它会把完整思考链（含敏感数据）打到日志。我们上线前加了日志脱敏中间件，过滤掉agent_scratchpad字段。

3.3 记忆（Memory）：别让Chatbot记住不该记的

ConversationBufferMemory是初学者最爱，也是线上事故最高发组件。某次银行项目上线后，客服机器人突然开始泄露前一位客户的卡号——因为ConversationBufferMemory默认把所有历史消息存在内存里，而Flask应用是多线程的。

LangChain的记忆模块本质是状态管理策略。我们为不同场景选型：

生产级记忆实现（PostgreSQL）：

from langchain_postgres import PostgresChatMessageHistory from langchain.memory import ConversationBufferMemory # 表结构已由langchain_postgres自动创建 history = PostgresChatMessageHistory( connection_string="postgresql://user:pass@localhost:5432/chatdb", table_name="message_store", session_id=f"{tenant_id}_{user_id}" # 会话粒度隔离 ) memory = ConversationBufferMemory( memory_key="chat_history", chat_memory=history, return_messages=True, input_key="input", output_key="response" ) # 在chain中使用 chain = LLMChain( llm=llm, prompt=prompt, memory=memory )

重要提醒：PostgresChatMessageHistory的session_id必须包含业务标识（如租户ID），否则所有用户消息会混在一个会话里。我们曾因此收到GDPR投诉——因为A用户的对话被B用户看到了。

4. 数据应用落地的七道生死关：从本地测试到K8s集群的避坑清单

4.1 第一道关：环境一致性——为什么你的本地能跑，线上必报错？

问题现象：本地Jupyter里langchain==0.1.16完美运行，部署到Airflow Docker镜像后，Chroma初始化失败。

根因分析：LangChain 0.1.x系列对底层依赖极其敏感。chromadb在0.4.22版本修复了SQLite并发锁，但langchain==0.1.16依赖的是chromadb<0.4.20。解决方案不是升级LangChain（会破坏现有链），而是锁定子依赖：

# Dockerfile FROM python:3.9-slim # 关键：显式安装兼容版本 RUN pip install \ langchain==0.1.16 \ chromadb==0.4.21 \ sentence-transformers==2.2.2 \ psycopg2-binary==2.9.7 COPY requirements.txt . RUN pip install -r requirements.txt

实操心得：永远用pip freeze > requirements-lock.txt生成锁文件。我们曾因pydantic从1.x升到2.x，导致所有PydanticOutputParser失效——因为v2的BaseModel构造函数签名变了。

4.2 第二道关：Token预算管理——别让LLM吃光你的GPU显存

问题现象：批量处理1000条数据时，第832条开始OOM，日志显示CUDA out of memory。

根因：LlamaCpp默认加载全部模型权重到GPU，而max_tokens参数只控制输出长度，不限制输入。解决方案是分片处理+显存监控：

from llama_cpp import Llama import torch # 初始化时指定GPU层 llm = Llama( model_path="./models/llama3-70b.Q4_K_M.gguf", n_gpu_layers=40, # 把前40层放GPU，其余放CPU n_ctx=2048, # 输入上下文限制 verbose=False ) # 批处理时动态计算token用量 def safe_batch_process(texts: list[str], batch_size: int = 8): for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] # 预估token数（用tiktoken粗略估算） total_tokens = sum(len(encoding.encode(t)) for t in batch) if total_tokens > 1500: # 留500token余量 # 拆更小的batch或截断 yield process_single(text) else: yield llm.create_chat_completion( messages=[{"role": "user", "content": t} for t in batch], max_tokens=256 )

经验：用tiktoken预估比实际llm.tokenize()快10倍，足够用于批处理调度。我们线上系统用Redis缓存各模型的token消耗率，动态调整batch_size。

4.3 第三道关：超时与重试——网络抖动不该让整个ETL失败

问题现象：调用OpenAI API时，偶尔超时导致Airflow Task失败，重试3次后仍失败。

标准解法是LangChain内置的RetryPolicy，但生产环境需要更精细控制：

from langchain_openai import ChatOpenAI from tenacity import retry, stop_after_attempt, wait_exponential # 自定义重试策略 @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10), reraise=True ) def robust_invoke(llm, prompt): return llm.invoke(prompt) # 在chain中封装 class RobustLLMChain(LLMChain): def _call(self, inputs: dict, run_manager=None) -> str: try: return robust_invoke(self.llm, self.prompt.format(**inputs)) except Exception as e: # 记录失败详情到监控系统 log_error("llm_invoke_failed", inputs, str(e)) raise # 使用 chain = RobustLLMChain(llm=llm, prompt=prompt)

关键：重试时必须传run_manager，否则LangChain的回调（CallbackHandler）会丢失。我们线上用Prometheus暴露llm_invoke_total和llm_invoke_failed_total指标，当失败率>5%时自动告警。

4.4 第四道关：输出稳定性——如何让LLM每次都说“人话”

问题现象：同一段SQL，LLM有时生成“查询成功”，有时生成“已执行SELECT语句”，导致下游解析失败。

解决方案不是调低temperature，而是用System Message固化输出契约：

from langchain_core.messages import SystemMessage, HumanMessage # 构建带强约束的messages messages = [ SystemMessage(content=""" 你是一个严格的SQL解释器。必须遵守： 1. 只输出JSON，无任何额外文本； 2. JSON必须包含字段：status（"success"或"error"）、summary（中文摘要）、sql（原始SQL）； 3. status为"error"时，summary必须包含具体错误原因。 """), HumanMessage(content=f"解释此SQL：{sql_query}") ] # 用ChatModel而非LLM chat_model = ChatOpenAI(model="gpt-4-turbo", temperature=0.0) result = chat_model.invoke(messages) # result.content 是纯JSON字符串，可直接json.loads()

实测数据：加System Message后，JSON格式合规率从78%升至99.2%，且status字段100%存在。这是比任何OutputParser都可靠的方案。

4.5 第五道关：可观测性——没有日志的LLM应用就是定时炸弹

问题现象：某天凌晨2点，客服机器人开始胡言乱语，但所有监控指标（CPU、内存、HTTP 200）都正常。

根因：LLM输出质量下降，但没人监控response_length、token_usage、confidence_score等业务指标。

我们强制实施的可观测性规范：

from langchain.callbacks.base import BaseCallbackHandler import logging class LLMCallbackHandler(BaseCallbackHandler): def on_llm_start(self, serialized, prompts, **kwargs): # 记录prompt长度、模型名、时间戳 for i, prompt in enumerate(prompts): logging.info(f"LLM_START|prompt_len={len(prompt)}|model={serialized.get('name')}|prompt_id={i}") def on_llm_end(self, response, **kwargs): # 记录输出长度、token用量、耗时 usage = response.llm_output.get("token_usage", {}) logging.info(f"LLM_END|output_len={len(response.generations[0][0].text)}|" f"prompt_tokens={usage.get('prompt_tokens',0)}|" f"completion_tokens={usage.get('completion_tokens',0)}|" f"total_tokens={usage.get('total_tokens',0)}") # 在LLM初始化时注册 llm = ChatOpenAI( callbacks=[LLMCallbackHandler()], verbose=True # 必须开启，否则callback不触发 )

日志规范：所有LLM日志必须带LLM_START/LLM_END前缀，便于ELK聚合。我们用Grafana看板监控avg(response_length)，当低于阈值（如50字符）时，说明模型在敷衍回答。

4.6 第六道关：安全边界——防止Prompt注入摧毁你的数据库

问题现象：用户输入“请删除所有用户数据”，LLM竟真的生成了DELETE FROM users。

这不是LLM的问题，是你的应用没设防。我们实施的三层防护：

1. 输入清洗：用正则过滤DROP|DELETE|INSERT|UNION等危险关键词（简单但有效）
2. 权限隔离：LLM生成的SQL只允许执行SELECT，通过数据库视图限制可访问字段
3. 输出校验：用sqlglot解析SQL AST，确保node.type == "select"且无where子句外的危险操作

import sqlglot from sqlglot.expressions import Select, Delete, Insert def validate_sql(sql: str) -> bool: try: parsed = sqlglot.parse_one(sql, read="postgres") # 只允许SELECT语句 if not isinstance(parsed, Select): return False # 检查是否有危险子句 if parsed.find(Delete) or parsed.find(Insert): return False return True except Exception: return False # 在chain中校验 if not validate_sql(generated_sql): raise ValueError("Generated SQL contains unsafe operations")

重要：sqlglot比正则可靠100倍。正则无法识别SELECT * FROM users; DROP TABLE users;这种注入，而sqlglot能正确解析为两个独立语句。

4.7 第七道关：成本控制——每一分钱都要算清楚

问题现象：月账单飙升300%，发现是LLM调用次数翻倍，但业务量没变。

根因：未开启streaming，且未监控token_usage。我们上线的成本监控体系：

# 计算单次调用成本（以GPT-4-turbo为例） def calculate_cost(prompt_tokens: int, completion_tokens: int) -> float: # $0.01/1K prompt tokens, $0.03/1K completion tokens return (prompt_tokens / 1000) * 0.01 + (completion_tokens / 1000) * 0.03 # 在callback中记录 class CostCallbackHandler(BaseCallbackHandler): def on_llm_end(self, response, **kwargs): usage = response.llm_output.get("token_usage", {}) cost = calculate_cost( usage.get("prompt_tokens", 0), usage.get("completion_tokens", 0) ) # 上报到Prometheus llm_cost_counter.labels(model="gpt-4-turbo").inc(cost) logging.info(f"LLM_COST|cost=${cost:.4f}|prompt_tokens={usage.get('prompt_tokens',0)}") # 设置预算告警 def check_budget(): today_cost = get_prometheus_metric("llm_cost_total", days=1) if today_cost > 500: # $500日预算 send_alert(f"LLM cost exceeded: ${today_cost:.2f}")

实操：我们给每个业务线分配独立model参数（如model="gpt-4-turbo-finance"），在Prometheus中按model标签分组计费，精确到分。

5. 真实世界问题排查手册：那些文档里绝不会写的故障现场

5.1 故障一：Chroma向量库查询结果为空，但count()显示有10万条数据

现象：vectorstore.similarity_search("iPhone")返回空列表，但vectorstore._collection.count()返回102456。

排查路径：

1. 检查embedding模型是否与索引时一致：HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")vsmodel_name="all-mpnet-base-v2"
2. 检查文本切分是否破坏语义："iPhone 15 Pro"被切成["iPhone 15", "Pro"]，导致无法匹配
3. 检查search_kwargs：k=1时可能因相似度阈值过高返回空，改为k=5, score_threshold=0.3

终极解法：用Chroma的get方法直接查ID，确认数据是否真存进去了：

# 查看前3条文档的ID和内容 docs = vectorstore._collection.get(limit=3) print([(d["id"], d["document"][:50]) for d in docs]) # 如果document是空字符串，说明DataFrameLoader的page_content_column指定错了

5.2 故障二：PydanticOutputParser抛OutputParserException，但错误信息不显示具体哪一行错

现象：parser.parse("{issues: ['a','b'], confidence: 0.95}")失败，错误信息只有“Failed to parse”而无位置。

根因：Pydantic v1的ValidationError默认不显示详细路径。解决方案是捕获异常并手动解析：

from pydantic import ValidationError try: result = parser.parse(text) except OutputParserException as e: # 重新用Pydantic解析，获取详细错误 try: ComplaintIssues.model_validate_json(text) except ValidationError as ve: # ve.errors()返回详细错误列表，含loc字段 for error in ve.errors(): print(f"Error at {error['loc']}: {error['msg']}") raise e

5.3 故障三：AgentExecutor在Airflow中无限循环，日志刷屏

现象：Agent反复调用同一个工具，如sql_search执行100次都不停。

根因：Agent的max_iterations默认是15，但某些LLM（如Phi-3）在工具调用失败时会重复发送相同指令。解决方案是强制限制：

agent_executor = AgentExecutor( agent=agent, tools=tools, max_iterations=5, # 严格限制 early_stopping_method="generate", # 失败时生成终止消息，而非重试 handle_parsing_errors=True # 自动处理解析错误 )

5.4 故障四：ConversationSummaryBufferMemory摘要丢失关键数字

现象：对话中提到“利率4.5%”，摘要后变成“利率约5%”。

根因：LLMChain用于摘要时，temperature=0.3导致数字被模糊化。解决方案是用temperature=0.0专用摘要链：

summary_prompt = PromptTemplate( template="请精准摘要以下对话，保留所有数字、日期、专有名词：\n{history}", input_variables=["history"] ) summary_chain = LLMChain( llm=ChatOpenAI(temperature=0.0), # 关键！ prompt=summary_prompt )

5.5 故障五：langchain==0.1.20升级后，RetrievalQAChain的return_source_documents=True不生效

现象：代码没变，但result["source_documents"]始终为空