用 LangChain 构建 LLM 工作流

**AI/LLM 应用开发实战系列** 第 3 篇

上一篇：[Prompt Engineering 核心技巧](./2026-02-27-llm-series-02.md) | 下一篇：RAG 检索增强生成

为什么需要 LangChain

直接使用 LLM API 时，你很快就会遇到这些问题：

1. **上下文管理复杂**：需要手动维护对话历史

2. **难以组合多个调用**：链式调用代码冗长

3. **缺少工具集成**：调用外部 API 需要自己封装

4. **难以测试和调试**：Prompt 散落在代码各处

LangChain 就是为了解决这些问题而生的框架。它提供了：

• **Chains**：组合多个 LLM 调用
• **Agents**：让 LLM 自主决定调用哪些工具
• **Memory**：自动管理对话历史
• **Tools**：预集成的常用工具（搜索、计算器、数据库等）

• 安装和配置

  
  pip install langchain langchain-openai langchain-community
  

  设置环境变量：

  
  export OPENAI_API_KEY="your-api-key"
  

  基础概念

  1. Prompt Templates

  Prompt 模板让你可以参数化 Prompt：

  
  from langchain.prompts import ChatPromptTemplate
  
  # 定义模板
  template = """
  你是一个{role}专家。请回答以下问题：
  
  问题：{question}
  """
  
  prompt = ChatPromptTemplate.from_template(template)
  
  # 使用模板
  messages = prompt.format_messages(
      role="Python 编程",
      question="什么是装饰器？"
  )
  
  print(messages[0].content)
  

  2. LLM Chain

  最简单的 Chain，将 Prompt 模板和 LLM 组合：

  
  from langchain_openai import ChatOpenAI
  from langchain.chains import LLMChain
  
  llm = ChatOpenAI(model="gpt-4", temperature=0.7)
  
  chain = LLMChain(
      llm=llm,
      prompt=prompt,
      output_key="answer"
  )
  
  # 调用
  result = chain.invoke({
      "role": "Python 编程",
      "question": "什么是装饰器？"
  })
  
  print(result["answer"])
  

  3. Sequential Chains

  按顺序执行多个 Chain：

  
  from langchain.chains import SequentialChain
  
  # 第一个 Chain：生成大纲
  outline_prompt = ChatPromptTemplate.from_template(
      "为关于{topic}的文章生成一个大纲，包含 5 个主要部分。"
  )
  outline_chain = LLMChain(llm=llm, prompt=outline_prompt, output_key="outline")
  
  # 第二个 Chain：根据大纲写引言
  intro_prompt = ChatPromptTemplate.from_template(
      "根据以下大纲写一篇引人入胜的引言：\n{outline}"
  )
  intro_chain = LLMChain(llm=llm, prompt=intro_prompt, output_key="introduction")
  
  # 组合成 SequentialChain
  sequential_chain = SequentialChain(
      chains=[outline_chain, intro_chain],
      input_variables=["topic"],
      output_variables=["outline", "introduction"],
      verbose=True
  )
  
  # 执行
  result = sequential_chain.invoke({"topic": "量子计算"})
  print(result["introduction"])
  

  实战：构建一个代码审查助手

  让我们构建一个完整的代码审查工作流：

  
  from langchain_openai import ChatOpenAI
  from langchain.prompts import ChatPromptTemplate
  from langchain.chains import SequentialChain, LLMChain
  
  llm = ChatOpenAI(model="gpt-4", temperature=0.3)
  
  # Step 1: 分析代码结构
  analyze_prompt = ChatPromptTemplate.from_template("""
  分析以下{language}代码的结构和主要功能：
  
  

  {code}

  
  
  请用 JSON 格式返回：
  {{
      "purpose": "代码的主要目的",
      "components": ["主要组件列表"],
      "complexity": "low/medium/high"
  }}
  """)
  analyze_chain = LLMChain(llm=llm, prompt=analyze_prompt, output_key="analysis")
  
  # Step 2: 识别潜在问题
  issues_prompt = ChatPromptTemplate.from_template("""
  基于以下代码分析，识别潜在的问题和改进建议：
  
  分析结果：{analysis}
  原始代码：{code}
  
  请列出：
  1. 代码质量问题（命名、结构等）
  2. 潜在 bug
  3. 性能问题
  4. 安全漏洞
  """)
  issues_chain = LLMChain(llm=llm, prompt=issues_prompt, output_key="issues")
  
  # Step 3: 生成改进建议
  suggestions_prompt = ChatPromptTemplate.from_template("""
  基于识别的问题，提供具体的改进建议和重构后的代码：
  
  问题列表：{issues}
  原始代码：{code}
  
  请提供：
  1. 优先修复的问题（按严重程度排序）
  2. 重构后的完整代码
  3. 改进说明
  """)
  suggestions_chain = LLMChain(llm=llm, prompt=suggestions_prompt, output_key="suggestions")
  
  # 组合成完整工作流
  code_review_chain = SequentialChain(
      chains=[analyze_chain, issues_chain, suggestions_chain],
      input_variables=["language", "code"],
      output_variables=["analysis", "issues", "suggestions"],
      verbose=True
  )
  
  # 使用示例
  code = """
  def calc(a,b):
      return a/b
  """
  
  result = code_review_chain.invoke({
      "language": "python",
      "code": code
  })
  
  print("=== 代码分析 ===")
  print(result["analysis"])
  print("\n=== 问题识别 ===")
  print(result["issues"])
  print("\n=== 改进建议 ===")
  print(result["suggestions"])
  

  使用 Memory 管理对话历史

  LangChain 的 Memory 模块可以自动管理对话历史：

  
  from langchain.memory import ConversationBufferMemory
  from langchain.chains import ConversationChain
  
  # 创建带记忆的对话链
  memory = ConversationBufferMemory()
  
  conversation = ConversationChain(
      llm=llm,
      memory=memory,
      verbose=True
  )
  
  # 多轮对话
  print(conversation.predict(input="你好，我叫小明"))
  print(conversation.predict(input="我喜欢编程"))
  print(conversation.predict(input="我刚才告诉你我叫什么？"))
  # LLM 会记住之前的对话内容
  

  使用 Agents 自主决策

  Agents 让 LLM 可以自主决定调用哪些工具：

  
  from langchain.agents import load_tools, initialize_agent, AgentType
  from langchain_community.tools import DuckDuckGoSearchRun
  
  # 初始化工具
  search = DuckDuckGoSearchRun()
  tools = [search]
  
  # 初始化 Agent
  agent = initialize_agent(
      tools=tools,
      llm=llm,
      agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
      verbose=True
  )
  
  # 让 Agent 自主决策
  result = agent.invoke("2024 年诺贝尔物理学奖得主是谁？")
  print(result["output"])
  

  最佳实践

  1. 合理设置 temperature

• 代码生成、数据分析：0.0-0.3
• 创意写作、头脑风暴：0.7-1.0

• 2. 使用缓存提高性能

  
  from langchain.cache import InMemoryCache
  from langchain.globals import set_llm_cache
  
  set_llm_cache(InMemoryCache())
  
  # 相同的 Prompt 会直接返回缓存结果
  

  3. 添加错误处理

  
  from langchain.callbacks import BaseCallbackHandler
  
  class ErrorHandler(BaseCallbackHandler):
      def on_llm_error(self, error, **kwargs):
          print(f"LLM 错误：{error}")
  
  llm.callbacks = [ErrorHandler()]
  

  总结

  LangChain 让 LLM 应用开发变得简单：

• ✅ **Chains**：组合多个 LLM 调用
• ✅ **Agents**：自主决策调用工具
• ✅ **Memory**：自动管理对话历史
• ✅ **Tools**：丰富的预集成工具

• 下一篇我们将学习 **RAG 检索增强生成**，让 LLM 拥有私有知识库。

  参考资源

• [LangChain 官方文档](https://python.langchain.com/)
• [LangChain GitHub](https://github.com/langchain-ai/langchain)
• [LangChain 中文教程](https://liaokong.gitbook.io/llm-kai-fa-jiao-cheng)