LangChain串联DeepSeek时，如何用自定义OutputParser解决‘思考污染’问题？

LangChain串联DeepSeek时如何用自定义OutputParser解决"思考污染"问题

当我们在LangChain框架中串联使用具备"思考过程"输出的推理模型（如DeepSeek）时，经常会遇到一个棘手的问题：前序节点的思考标签会污染后续节点的提示词，导致整个链路的逻辑混乱。本文将深入探讨这一问题的成因，并通过一个"大象塞冰箱"的趣味案例，展示如何设计一个"两步走"的自定义OutputParser来彻底解决这个问题。

1. 问题现象与诊断

让我们从一个经典的"大象塞冰箱"案例开始，直观感受"思考污染"带来的问题。假设我们构建了一个包含三个步骤的工作流：

1. 打开冰箱
2. 把大象塞进冰箱
3. 关闭冰箱

当使用标准的LangChain串联方式时，我们会发现一个奇怪的现象：每个步骤的输出都包含了前序步骤的思考过程，导致后续步骤的提示词被污染。具体表现为：

{'step_1': '<think>...思考过程...</think>打开冰箱的方法...', 'step_2': '<think>...思考过程...</think>打开冰箱的方法...', # 被污染 'step_3': '<think>...思考过程...</think>打开冰箱的方法...'} # 被污染

这种污染会导致整个工作流偏离预期，所有节点都在重复第一个节点的思考过程。问题的根源在于：

• 推理模型（如DeepSeek）会输出带有<think>标签的思考过程
• 这些思考过程会被传递到后续节点的提示词中
• 后续节点会基于被污染的提示词继续生成内容

2. 解决方案设计

要解决这个问题，我们需要设计一个能够"净化"输出的自定义OutputParser。这个解析器需要完成两个关键任务：

1. 剥离<think>标签及其内容
2. 提取<answer>标签中的最终答案

以下是解决方案的核心思路：

2.1 引导模型结构化输出

首先，我们需要修改提示词模板，明确要求模型将输出内容放在<answer>标签中：

prompt = PromptTemplate.from_template( "你是一名厨师，怎么打开冰箱？" "输出内容放在<answer></answer>之间" )

2.2 实现两步解析器

接下来，我们实现一个自定义的DoubleStepOutputParser：

import re class DoubleStepOutputParser(StrOutputParser): """专用输出解析器，分步处理标签""" def parse(self, text: str) -> str: # 第一步：删除所有<think>标签及内容（包括跨行情况） cleaned_text = re.sub( r'<think>.*?</think>', # 非贪婪匹配 '', text, flags=re.DOTALL # 支持跨行匹配 ) # 第二步：提取<answer>内容 answer_match = re.search( r'<answer>(.*?)</answer>', cleaned_text, re.DOTALL ) return answer_match.group(1).strip() if answer_match else cleaned_text

这个解析器的工作原理是：

1. 使用正则表达式删除所有<think>标签及其内容
2. 从剩余文本中提取<answer>标签内的内容
3. 如果找不到<answer>标签，则返回清理后的文本

3. 完整实现方案

让我们将上述组件整合到一个完整的工作流中：

def elephant_stuffed_into_refrigerator_solution(): """解决方案：使用自定义OutputParser""" model = ChatOpenAI( model="DS70B", base_url="YOURS", api_key="EMPTY" ) # 自定义解析器实例 output_parser = DoubleStepOutputParser() # Chain 1：打开冰箱 prompt_symptom = PromptTemplate.from_template( "你是一名厨师，怎么打开冰箱？" "输出内容放在<answer></answer>之间" ) chain_one = ( prompt_symptom | model | output_parser ).with_config(output_key="step_1") # Chain2：把大象塞进冰箱 prompt_diagnosis = PromptTemplate.from_template( "目前已经完成{step_1}内容" "你是一名厨师，怎么把大象塞进冰箱？" "输出内容放在<answer></answer>之间" ) chain_two = ( prompt_diagnosis | model | output_parser ).with_config(output_key="step_2") # Chain3：关闭冰箱 prompt_diag_extract = PromptTemplate.from_template( "目前已经完成{step_2}内容" "你是一名厨师，怎么关闭冰箱？" "输出内容放在<answer></answer>之间" ) chain_three = ( prompt_diag_extract | model | output_parser ).with_config(output_key="step_3") # 组合处理链 overall_chain = ( {"project_desc": RunnablePassthrough()} | RunnablePassthrough.assign(step_1=chain_one) | RunnablePassthrough.assign(step_2=chain_two) | RunnablePassthrough.assign(step_3=chain_three) ) final_res = overall_chain.invoke("") return { "step_1": final_res["step_1"], "step_2": final_res["step_2"], "step_3": final_res["step_3"] }

4. 效果对比与最佳实践

使用自定义OutputParser前后的效果对比：

在实际应用中，我们还需要注意以下几点：

1. 正则表达式优化：根据模型输出的具体格式调整正则表达式，确保能正确匹配各种格式的标签
2. 错误处理：增强解析器的鲁棒性，处理各种可能的异常情况
3. 性能考虑：对于大规模工作流，可以考虑缓存解析结果

# 增强版的错误处理 class RobustDoubleStepOutputParser(StrOutputParser): def parse(self, text: str) -> str: try: # 删除<think>标签 cleaned_text = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL) # 提取<answer>内容 answer_match = re.search(r'<answer>(.*?)</answer>', cleaned_text, re.DOTALL) if answer_match: return answer_match.group(1).strip() # 如果没有<answer>标签，尝试其他可能的标签 for tag in ['<response>', '<output>', '<result>']: match = re.search(fr'{tag}(.*?){tag.replace("<", "</")}', cleaned_text, re.DOTALL) if match: return match.group(1).strip() return cleaned_text.strip() except Exception as e: print(f"解析错误: {e}") return text # 返回原始文本作为后备

5. 高级应用场景

这种自定义OutputParser的技术不仅适用于简单的"大象塞冰箱"案例，还可以应用于更复杂的场景：

1. 多步骤决策系统：确保每个决策步骤的输出不会被前序步骤的思考过程污染
2. 状态保持工作流：在需要保持状态的长时间对话中，清理中间思考过程
3. 复杂任务分解：将大任务分解为多个子任务时，保持每个子任务的独立性

以下是一个更复杂的应用示例，展示如何在多步骤数据分析工作流中使用这项技术：

def data_analysis_workflow(): """多步骤数据分析工作流""" model = ChatOpenAI(model="DS70B") parser = RobustDoubleStepOutputParser() # 步骤1：数据加载 prompt_load = PromptTemplate.from_template( "加载数据集{dataset_path}并执行初步检查" "输出放在<answer></answer>之间" ) chain_load = (prompt_load | model | parser).with_config(output_key="load_result") # 步骤2：数据清洗 prompt_clean = PromptTemplate.from_template( "基于{load_result}，执行数据清洗" "输出放在<answer></answer>之间" ) chain_clean = (prompt_clean | model | parser).with_config(output_key="clean_result") # 步骤3：分析建模 prompt_analyze = PromptTemplate.from_template( "基于{clean_result}，执行分析建模" "输出放在<answer></answer>之间" ) chain_analyze = (prompt_analyze | model | parser).with_config(output_key="analysis_result") # 组合工作流 workflow = ( {"dataset_path": RunnablePassthrough()} | RunnablePassthrough.assign(load_result=chain_load) | RunnablePassthrough.assign(clean_result=chain_clean) | RunnablePassthrough.assign(analysis_result=chain_analyze) ) return workflow.invoke("sales_data.csv")

在这个示例中，自定义OutputParser确保了每个步骤的输出都是干净的，不会被前序步骤的思考过程污染，从而保证了整个分析工作流的正确性。