案例实战LangChain的核武器-chain

本文将介绍LangChain框架中最核心的Chain组件，通过三个渐进式案例展示如何构建高效的大语言模型工作流。我们将从简单的链式组合开始，逐步深入到路由逻辑和代码生成检查的复杂场景。

1. 链式组合（Sequential Chains）

顺序执行链
本质：函数式编程的compose操作
核心：数据流单向管道（A→B→C）
一句话：“LLM版的UNIX管道”

案例：文本总结+翻译组合链

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI# 初始化模型
llm = ChatOpenAI(base_url="https://api.deepseek.com/", api_key="your-api-key",model="deepseek-chat")# 1. 总结链
summary_chain = (ChatPromptTemplate.from_template("用一句话总结：{text}")| llm
)# 2. 翻译链
translation_chain = (ChatPromptTemplate.from_template("翻译成中文：{text}")| llm
)# 组合链（管道操作）
report_chain = summary_chain | translation_chain# 执行
english_text = "LangChain provides tools to build LLM-powered applications."
result = report_chain.invoke({"text": english_text})
print(result.content)

关键点：

• 使用|管道符连接多个链
• 前一个链的输出自动作为下一个链的输入
• 适合线性任务流（如：总结→翻译→格式化）

---

2. 路由链（Router Chains）

选择链
本质：模式匹配的状态机
核心：(输入→谓词)→目标链的映射
一句话：“LLM流量控制器”

案例：领域专家路由系统

from langchain_core.runnables import RunnableBranch# 定义领域专家链
physics_chain = (ChatPromptTemplate.from_template("你是一个物理学家，请专业回答：{input}")| llm
)math_chain = (ChatPromptTemplate.from_template("你是一个数学家，请分步骤解答：{input}")| llm
)general_chain = (ChatPromptTemplate.from_template("回答这个问题：{input}")| llm
)# 构建路由逻辑
router_chain = RunnableBranch((lambda x: "物理" in x["input"], physics_chain),(lambda x: "数学" in x["input"], math_chain),general_chain  # 默认链
)# 测试
questions = [{"input": "解释相对论"},      # 路由到物理链{"input": "解方程x^2=4"},    # 路由到数学链{"input": "如何煮咖啡"}       # 路由到通用链
]
for q in questions:print(router_chain.invoke(q))

关键点：

• RunnableBranch实现条件路由
• 每个分支可定制不同的prompt模板
• 适合多场景分流（如客服、教育领域）

---

3. 代码生成与验证链

让chain执行自定义接口
本质：编译器前端简化版
核心：生成→解析→沙箱三级流水线
一句话：“AI时代的编译器”

案例：AI代码生成+语法检查

from langchain_core.runnables import RunnableLambda
import ast, re# 代码生成链
code_chain = (ChatPromptTemplate.from_template("用Python实现：{task}")| llm| {"code": RunnablePassthrough()}  # 保留原始输出
)# 语法检查函数
def syntax_check(code_dict):try:content = code_dict["code"].contentcode_match = re.search(r'```python\n(.*?)\n```', content, re.DOTALL)if code_match:code = code_match.group(1)ast.parse(code)  # 语法验证return {"status": "VALID", "code": code}return {"status": "INVALID: No code block"}except SyntaxError as e:return {"status": f"INVALID: {e}"}# 组合链
full_chain = code_chain | RunnableLambda(syntax_check)# 测试
result = full_chain.invoke({"task": "快速排序函数"})
if result["status"] == "VALID":print("生成的代码：\n", result["code"])
else:print("错误：", result["status"])

关键点：

• RunnableLambda实现自定义处理逻辑
• 正则表达式提取Markdown代码块
• ast.parse进行Python语法验证
• 适合代码生成类场景的完整流水线

---

对比总结

功能类型	适用场景	核心组件	案例亮点
链式组合	线性多步骤任务	管道符`	`
路由链	多分支条件逻辑	RunnableBranch	领域专家路由
验证链	输入/输出处理与校验	RunnableLambda	代码生成+语法检查

---