LangChain 框架
LangChain 是目前最流行的 Agent 开发框架，由 Harrison Chase 于 2022 年创立。它提供了构建 LLM 应用所需的全套组件，包括模型封装、提示词管理、链式调用、工具集成和记忆系统等。
一、核心原理
1.1 LangChain 设计哲学
LangChain 的设计理念是"组合优先"（Composition First）：
┌─────────────────────────────────────────────────────────────┐
│ LangChain 设计哲学 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 核心思想：像搭积木一样构建 LLM 应用 │
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ LLM │ + │ Prompt │ + │ Tool │ + │ Memory │ │
│ │ 组件 │ │ 模板 │ │ 工具 │ │ 记忆 │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │ │
│ └─────────────┼─────────────┼─────────────┘ │
│ │ │ │
│ ↓ ↓ │
│ ┌─────────────────────────────┐ │
│ │ Chain │ │
│ │ (链式组合) │ │
│ └─────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────┐ │
│ │ Agent │ │
│ │ (智能决策层) │ │
│ └─────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 核心组件架构
┌─────────────────────────────────────────────────────────────────────┐
│ LangChain 核心组件架构 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 应用层 (Application) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Agent │ │ Chain │ │ RAG App │ │ Chat App │ │ │
│ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │
│ └───────┼─────────────┼─────────────┼─────────────┼───────────┘ │
│ │ │ │ │ │
│ ┌───────┴─────────────┴─────────────┴─────────────┴───────────┐ │
│ │ 核心层 (Core) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Prompt │ │ Output │ │ Memory │ │Document │ │ │
│ │ │ Template│ │ Parser │ │ │ │ Loader │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │ │ │ │ │
│ ┌───────┴─────────────┴─────────────┴─────────────┴───────────┐ │
│ │ 集成层 (Integration) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ LLM │ │ Tools │ │ Vector │ │ Other │ │ │
│ │ │ Providers│ │ │ │ Stores │ │ Services │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │
│ └──────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘1.3 核心概念详解
| 概念 | 英文 | 作用 | 示例 |
|---|---|---|---|
| Chain | 链 | 将多个组件串联成工作流 | LLMChain, SequentialChain |
| Agent | 智能体 | 动态决策和工具调用 | ReActAgent, OpenAIFunctionsAgent |
| Tool | 工具 | 外部能力集成 | Search, Calculator, PythonREPL |
| Memory | 记忆 | 对话历史管理 | ConversationBufferMemory |
| Prompt | 提示词 | 输入模板管理 | PromptTemplate, ChatPromptTemplate |
| Output Parser | 输出解析 | 结构化输出提取 | PydanticOutputParser, JsonOutputParser |
二、核心组件详解
2.1 Model I/O（模型输入输出）
Model I/O 是 LangChain 与 LLM 交互的基础层：
┌─────────────────────────────────────────────────────────────┐
│ Model I/O 架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 输入处理 模型调用 输出处理 │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Prompt │ ─────→ │ LLM │ ──────→ │ Output │ │
│ │Template │ │ Model │ │ Parser │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 格式化 │ │ API调用 │ │ 结构化 │ │
│ │ 变量注入│ │ 响应获取│ │ 数据提取│ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘代码示例：
"""
LangChain Model I/O 示例
演示 Prompt Template、LLM 和 Output Parser 的使用
"""
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List
# 1. 定义输出结构
class MovieReview(BaseModel):
"""电影评论结构"""
title: str = Field(description="电影名称")
rating: int = Field(description="评分 1-10")
summary: str = Field(description="一句话总结")
pros: List[str] = Field(description="优点列表")
cons: List[str] = Field(description="缺点列表")
# 2. 创建输出解析器
parser = PydanticOutputParser(pydantic_object=MovieReview)
# 3. 创建提示词模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一位专业的影评人。请按照指定格式输出影评。"),
("human", "{query}"),
("system", "{format_instructions}")
])
# 4. 创建 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 5. 构建链
chain = prompt | llm | parser
# 6. 执行
result = chain.invoke({
"query": "请评价电影《盗梦空间》",
"format_instructions": parser.get_format_instructions()
})
print(f"电影: {result.title}")
print(f"评分: {result.rating}/10")
print(f"总结: {result.summary}")
print(f"优点: {result.pros}")
print(f"缺点: {result.cons}")2.2 Chain（链）
Chain 是 LangChain 的核心概念，用于将多个组件串联：
┌─────────────────────────────────────────────────────────────┐
│ Chain 类型全景 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ LLMChain (基础链) │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Prompt │ → │ LLM │ → │ Output │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ SequentialChain (顺序链) │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Chain 1 │ → │ Chain 2 │ → │ Chain 3 │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ RouterChain (路由链) │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ Router (路由器) │ │ │
│ │ └─────────────────┬───────────────────────────┘ │ │
│ │ ┌────────┼────────┐ │ │
│ │ ↓ ↓ ↓ │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Chain A │ │ Chain B │ │ Chain C │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ TransformChain (转换链) │ │
│ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │
│ │ │ Input │ → │Transform│ → │ Output │ │ │
│ │ │ │ │ Function│ │ │ │ │
│ │ └─────────┘ └─────────┘ └─────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘LCEL（LangChain Expression Language）示例：
"""
使用 LCEL 构建链
LCEL 是 LangChain 2.0 引入的声明式语法，使用 | 操作符连接组件
"""
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser
from langchain.schema.runnable import RunnableParallel, RunnablePassthrough
# 1. 创建组件
model = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template(
"请用简洁的语言回答：{question}"
)
output_parser = StrOutputParser()
# 2. 使用 LCEL 构建简单链
simple_chain = prompt | model | output_parser
# 3. 执行简单链
result = simple_chain.invoke({"question": "什么是量子计算？"})
print(result)
# 4. 使用 RunnableParallel 构建并行链
parallel_chain = RunnableParallel(
summary=prompt | model | output_parser,
original=RunnablePassthrough()
)
# 5. 执行并行链
result = parallel_chain.invoke({"question": "什么是区块链？"})
print(f"原始问题: {result['original']}")
print(f"回答: {result['summary']}")
# 6. 构建复杂链（带重试和回调）
from langchain.schema.runnable import RunnableRetry
robust_chain = (
prompt
| model
| output_parser
).with_retry(
stop_after_attempt=3,
wait_exponential_multiplier=1000
).with_fallbacks([
ChatOpenAI(model="gpt-3.5-turbo") | output_parser
])2.3 Agent（智能体）
Agent 是 LangChain 中最强大的组件，能够动态决策和调用工具：
┌─────────────────────────────────────────────────────────────┐
│ Agent 工作流程 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 用户输入 │ │
│ └──────┬──────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Agent Loop (循环) │ │
│ │ ┌───────────────────────────────────────────────┐ │ │
│ │ │ 1. 思考 (Think) │ │ │
│ │ │ ┌─────────────────────────────────────────┐ │ │ │
│ │ │ │ 分析输入，决定下一步行动 │ │ │ │
│ │ │ └─────────────────────────────────────────┘ │ │ │
│ │ └───────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌───────────────────────────────────────────────┐ │ │
│ │ │ 2. 行动 (Act) │ │ │
│ │ │ ┌─────────────────────────────────────────┐ │ │ │
│ │ │ │ 选择工具，传入参数，执行调用 │ │ │ │
│ │ │ └─────────────────────────────────────────┘ │ │ │
│ │ └───────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌───────────────────────────────────────────────┐ │ │
│ │ │ 3. 观察 (Observe) │ │ │
│ │ │ ┌─────────────────────────────────────────┐ │ │ │
│ │ │ │ 获取工具执行结果，更新上下文 │ │ │ │
│ │ │ └─────────────────────────────────────────┘ │ │ │
│ │ └───────────────────────────────────────────────┘ │ │
│ │ │ │ │
│ │ ┌────────┴────────┐ │ │
│ │ ↓ ↓ │ │
│ │ [继续循环] [生成答案] │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────┐ │
│ │ 最终输出 │ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘Agent 类型对比：
| Agent 类型 | 特点 | 适用场景 |
|---|---|---|
| ZeroShotAgent | 零样本推理，无示例 | 通用场景 |
| ReActAgent | 思考-行动-观察循环 | 需要工具调用的场景 |
| OpenAIFunctionsAgent | 使用 OpenAI Function Calling | OpenAI 模型最佳实践 |
| StructuredChatAgent | 支持多输入工具 | 复杂工具调用场景 |
| ConversationalAgent | 优化对话体验 | 聊天机器人场景 |
代码示例：
"""
LangChain Agent 示例
创建一个具备搜索和计算能力的 Agent
"""
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
from langchain import hub
# 1. 定义工具
def search_tool(query: str) -> str:
"""搜索工具（模拟）"""
# 实际应用中调用真实搜索 API
mock_data = {
"天气": "北京今天晴天，气温 15-25°C",
"新闻": "今日科技新闻：AI 技术持续突破",
"股票": "苹果股价今日收盘 $178.50"
}
for key, value in mock_data.items():
if key in query:
return value
return f"未找到关于 '{query}' 的信息"
def calculator_tool(expression: str) -> str:
"""计算器工具"""
try:
result = eval(expression)
return f"计算结果: {result}"
except Exception as e:
return f"计算错误: {str(e)}"
def weather_tool(city: str) -> str:
"""天气查询工具"""
# 模拟天气 API
weather_data = {
"北京": "晴天，15-25°C，空气质量良好",
"上海": "多云，18-28°C，有轻微雾霾",
"广州": "小雨，22-30°C，湿度较高"
}
return weather_data.get(city, f"未找到 {city} 的天气信息")
# 2. 创建工具列表
tools = [
Tool(
name="search",
func=search_tool,
description="搜索互联网获取信息，输入搜索关键词"
),
Tool(
name="calculator",
func=calculator_tool,
description="执行数学计算，输入数学表达式，如 '2+2' 或 '10*5'"
),
Tool(
name="weather",
func=weather_tool,
description="查询城市天气，输入城市名称"
)
]
# 3. 创建 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 4. 获取提示词模板
prompt = hub.pull("hwchase17/openai-functions-agent")
# 5. 创建 Agent
agent = create_openai_functions_agent(
llm=llm,
tools=tools,
prompt=prompt
)
# 6. 创建 Agent 执行器
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
verbose=True, # 打印详细执行过程
max_iterations=10, # 最大迭代次数
handle_parsing_errors=True, # 处理解析错误
max_execution_time=60 # 最大执行时间（秒）
)
# 7. 执行 Agent
result = agent_executor.invoke({
"input": "北京今天天气怎么样？如果温度适中，计算一下 15+10 的结果"
})
print(f"\n最终答案: {result['output']}")2.4 Memory（记忆系统）
Memory 用于管理对话历史和上下文：
┌─────────────────────────────────────────────────────────────┐
│ Memory 类型架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ ConversationBufferMemory (缓冲区记忆) │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ 存储完整对话历史 │ │ │
│ │ │ User: 你好 → AI: 你好！有什么可以帮助你的？ │ │ │
│ │ │ User: 天气 → AI: 请问您想查询哪个城市？ │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ │ 特点：完整保留，Token 消耗大 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ ConversationBufferWindowMemory (窗口记忆) │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ 只保留最近 k 轮对话 │ │ │
│ │ │ [最近3轮] │ │ │
│ │ │ User: x → AI: y (最近) │ │ │
│ │ │ User: x → AI: y (第2近) │ │ │
│ │ │ User: x → AI: y (第3近) │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ │ 特点：控制长度，可能丢失早期上下文 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ ConversationSummaryMemory (摘要记忆) │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ 压缩历史对话为摘要 │ │ │
│ │ │ 摘要：用户询问了北京天气，AI推荐了户外活动 │ │ │
│ │ │ 当前：用户询问明天天气... │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ │ 特点：节省 Token，但摘要可能丢失细节 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ VectorStoreMemory (向量记忆) │ │
│ │ ┌─────────────────────────────────────────────┐ │ │
│ │ │ 向量数据库存储 + 相似性检索 │ │ │
│ │ │ [向量存储] → [相关性检索] → [检索结果] │ │ │
│ │ └─────────────────────────────────────────────┘ │ │
│ │ 特点：支持大规模历史，检索相关记忆 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘代码示例：
"""
LangChain Memory 示例
演示不同类型记忆系统的使用
"""
from langchain_openai import ChatOpenAI
from langchain.memory import (
ConversationBufferMemory,
ConversationBufferWindowMemory,
ConversationSummaryMemory
)
from langchain.chains import ConversationChain
# 1. ConversationBufferMemory（完整缓冲）
buffer_memory = ConversationBufferMemory(
memory_key="history",
return_messages=True
)
# 2. ConversationBufferWindowMemory（滑动窗口）
window_memory = ConversationBufferWindowMemory(
k=3, # 保留最近 3 轮对话
memory_key="history",
return_messages=True
)
# 3. ConversationSummaryMemory（摘要压缩）
llm = ChatOpenAI(model="gpt-4")
summary_memory = ConversationSummaryMemory(
llm=llm,
memory_key="history",
return_messages=True
)
# 4. 创建对话链
conversation = ConversationChain(
llm=llm,
memory=buffer_memory,
verbose=True
)
# 5. 进行对话
response1 = conversation.predict(input="你好，我是小明")
print(f"AI: {response1}")
response2 = conversation.predict(input="我喜欢打篮球")
print(f"AI: {response2}")
response3 = conversation.predict(input="你记得我叫什么名字吗？")
print(f"AI: {response3}")
# 6. 查看记忆内容
print(f"\n对话历史：\n{buffer_memory.load_memory_variables({})}")2.5 Tools（工具系统）
Tools 是 Agent 与外部世界交互的桥梁：
┌─────────────────────────────────────────────────────────────┐
│ Tools 系统架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Built-in Tools (内置工具) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Search │ │ Wikipedia│ │ Python │ │ │
│ │ │ │ │ │ │ REPL │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Requests │ │ Terminal │ │ Shell │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Custom Tools (自定义工具) │ │
│ │ │ │
│ │ class MyTool(BaseTool): │ │
│ │ name = "my_tool" │ │
│ │ description = "工具描述" │ │
│ │ │ │
│ │ def _run(self, query: str) -> str: │ │
│ │ # 工具逻辑 │ │
│ │ return result │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Tool Integration (工具集成) │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Zapier │ │ OpenAPI │ │ SQL │ │ │
│ │ │ NLA │ │ Spec │ │ Database │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘自定义工具示例：
"""
自定义工具示例
展示如何创建符合 LangChain 规范的工具
"""
from typing import Optional, Type
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
# 1. 定义输入参数模型
class StockPriceInput(BaseModel):
"""股票价格查询输入"""
symbol: str = Field(description="股票代码，如 AAPL, GOOGL")
date: Optional[str] = Field(
default=None,
description="查询日期，格式 YYYY-MM-DD，默认今天"
)
# 2. 定义工具类
class StockPriceTool(BaseTool):
"""股票价格查询工具"""
name = "stock_price"
description = "查询股票价格信息"
args_schema: Type[BaseModel] = StockPriceInput
def _run(
self,
symbol: str,
date: Optional[str] = None
) -> str:
"""执行工具逻辑"""
# 模拟股票数据
stock_data = {
"AAPL": {"price": 178.50, "change": "+2.30"},
"GOOGL": {"price": 141.80, "change": "-1.20"},
"MSFT": {"price": 378.90, "change": "+5.60"}
}
symbol = symbol.upper()
if symbol not in stock_data:
return f"未找到股票代码 '{symbol}'"
data = stock_data[symbol]
return (
f"股票 {symbol}:\n"
f" 当前价格: ${data['price']}\n"
f" 涨跌幅: {data['change']}"
)
async def _arun(
self,
symbol: str,
date: Optional[str] = None
) -> str:
"""异步执行（可选）"""
return self._run(symbol, date)
# 3. 使用装饰器创建工具（简化方式）
from langchain.tools import tool
@tool
def calculate_mortgage(
principal: float,
rate: float,
years: int
) -> str:
"""
计算房贷月供
Args:
principal: 贷款本金（元）
rate: 年利率（如 0.05 表示 5%）
years: 贷款年限
Returns:
月供金额和总利息信息
"""
monthly_rate = rate / 12
months = years * 12
# 等额本息公式
monthly_payment = principal * (
monthly_rate * (1 + monthly_rate) ** months
) / ((1 + monthly_rate) ** months - 1)
total_payment = monthly_payment * months
total_interest = total_payment - principal
return (
f"贷款信息:\n"
f" 月供: ¥{monthly_payment:,.2f}\n"
f" 总还款: ¥{total_payment:,.2f}\n"
f" 总利息: ¥{total_interest:,.2f}"
)
# 4. 使用工具
tools = [StockPriceTool(), calculate_mortgage]
# 创建 Agent 使用这些工具
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain import hub
llm = ChatOpenAI(model="gpt-4")
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 执行
result = agent_executor.invoke({
"input": "查询苹果公司(AAPL)的股价，然后计算贷款100万、利率4.9%、30年的月供"
})
print(result["output"])三、RAG 应用实现
3.1 RAG 架构概述
┌─────────────────────────────────────────────────────────────┐
│ RAG 完整架构 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 离线索引构建阶段 │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Document │ → │ Split │ → │ Embedding│ │ │
│ │ │ Loader │ │ 分块 │ │ 向量化 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌──────────┐ │ │
│ │ │ Vector │ │ │
│ │ │ Store │ │ │
│ │ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 在线查询阶段 │ │
│ │ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Query │ → │ Embedding│ → │Retrieve │ │ │
│ │ │ 问题 │ │ 向量化 │ │ 检索 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ LLM │ ← │ Prompt │ ← │ Context │ │ │
│ │ │ 生成 │ │ 构建 │ │ 组装 │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ │ │ │ │
│ │ ↓ │ │
│ │ ┌──────────┐ │ │
│ │ │ Answer │ │ │
│ │ │ 回答 │ │ │
│ │ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.2 RAG 完整实现
"""
LangChain RAG 完整实现示例
构建一个基于文档的问答系统
"""
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import (
PyPDFLoader,
TextLoader,
DirectoryLoader
)
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# ========== 1. 文档加载 ==========
def load_documents(data_path: str):
"""加载文档"""
# 加载 PDF 文件
pdf_loader = DirectoryLoader(
data_path,
glob="**/*.pdf",
loader_cls=PyPDFLoader
)
# 加载文本文件
txt_loader = DirectoryLoader(
data_path,
glob="**/*.txt",
loader_cls=TextLoader
)
documents = pdf_loader.load() + txt_loader.load()
return documents
# ========== 2. 文档分割 ==========
def split_documents(documents, chunk_size=1000, chunk_overlap=200):
"""分割文档为小块"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
length_function=len,
separators=["\n\n", "\n", " ", ""]
)
chunks = text_splitter.split_documents(documents)
return chunks
# ========== 3. 向量存储 ==========
def create_vector_store(chunks, persist_directory="./chroma_db"):
"""创建向量存储"""
embeddings = OpenAIEmbeddings()
vector_store = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory=persist_directory
)
return vector_store
# ========== 4. 检索器配置 ==========
def create_retriever(vector_store, search_type="mmr", k=4):
"""创建检索器"""
# 基础检索器
base_retriever = vector_store.as_retriever(
search_type=search_type,
search_kwargs={"k": k}
)
# 使用 LLM 进行上下文压缩
llm = ChatOpenAI(model="gpt-4", temperature=0)
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=base_retriever
)
return compression_retriever
# ========== 5. RAG 链构建 ==========
def create_rag_chain(retriever):
"""创建 RAG 问答链"""
# 自定义提示词模板
prompt_template = """
你是一个专业的问答助手。请根据以下上下文回答问题。
如果上下文中没有相关信息，请说"根据已有资料无法回答该问题"。
上下文：
{context}
问题：{question}
请提供详细、准确的回答：
"""
PROMPT = PromptTemplate(
template=prompt_template,
input_variables=["context", "question"]
)
# 创建 LLM
llm = ChatOpenAI(model="gpt-4", temperature=0)
# 创建问答链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True,
chain_type_kwargs={"prompt": PROMPT}
)
return qa_chain
# ========== 6. 完整 RAG 应用 ==========
class RAGApplication:
"""RAG 应用类"""
def __init__(self, data_path: str, persist_directory: str = "./chroma_db"):
"""初始化 RAG 应用"""
self.data_path = data_path
self.persist_directory = persist_directory
# 初始化组件
self.vector_store = None
self.retriever = None
self.qa_chain = None
def build_index(self):
"""构建索引"""
print("1. 加载文档...")
documents = load_documents(self.data_path)
print(f" 加载了 {len(documents)} 个文档")
print("2. 分割文档...")
chunks = split_documents(documents)
print(f" 分割为 {len(chunks)} 个文本块")
print("3. 创建向量存储...")
self.vector_store = create_vector_store(
chunks,
self.persist_directory
)
print(" 向量存储创建完成")
def load_index(self):
"""加载已有索引"""
embeddings = OpenAIEmbeddings()
self.vector_store = Chroma(
persist_directory=self.persist_directory,
embedding_function=embeddings
)
def setup(self, rebuild=False):
"""设置应用"""
if rebuild:
self.build_index()
else:
self.load_index()
print("4. 创建检索器...")
self.retriever = create_retriever(self.vector_store)
print("5. 创建问答链...")
self.qa_chain = create_rag_chain(self.retriever)
print("RAG 应用准备就绪！")
def query(self, question: str) -> dict:
"""查询问题"""
result = self.qa_chain.invoke({"query": question})
return {
"answer": result["result"],
"sources": [
doc.metadata.get("source", "未知来源")
for doc in result["source_documents"]
]
}
# ========== 使用示例 ==========
if __name__ == "__main__":
# 创建应用
app = RAGApplication(
data_path="./documents",
persist_directory="./chroma_db"
)
# 设置（首次运行需要 rebuild=True）
app.setup(rebuild=False)
# 查询
result = app.query("什么是机器学习？")
print(f"\n回答：{result['answer']}")
print(f"\n来源：{result['sources']}")四、生产最佳实践
4.1 架构设计原则
┌─────────────────────────────────────────────────────────────┐
│ 生产级架构设计 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ API 层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ FastAPI │ │ Rate │ │ Auth │ │ │
│ │ │ Endpoint │ │ Limiter │ │ Guard │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 服务层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Agent │ │ Chain │ │ Memory │ │ │
│ │ │ Service │ │ Service │ │ Service │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 数据层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Vector │ │ Cache │ │ DB │ │ │
│ │ │ Store │ │ (Redis) │ │ (Postgres)│ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 可观测层 │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │ Logging │ │ Metrics │ │ Tracing │ │ │
│ │ │ │ │ (Prom) │ │ (Jaeger) │ │ │
│ │ └──────────┘ └──────────┘ └──────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘4.2 错误处理与重试
"""
生产级错误处理和重试策略
"""
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type
)
from langchain.schema import OutputParserException
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RobustChain:
"""具有错误处理能力的链"""
def __init__(self, chain, max_retries=3):
self.chain = chain
self.max_retries = max_retries
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type(Exception),
before_sleep=lambda retry_state: logger.warning(
f"重试 {retry_state.attempt_number}..."
)
)
def invoke(self, input_data):
"""执行链，带重试"""
try:
return self.chain.invoke(input_data)
except OutputParserException as e:
logger.error(f"输出解析错误: {e}")
# 尝试修复输出
return self._handle_parse_error(e, input_data)
except Exception as e:
logger.error(f"执行错误: {e}")
raise
def _handle_parse_error(self, error, input_data):
"""处理解析错误"""
# 实现修复逻辑
logger.info("尝试修复输出格式...")
# 可以添加格式修复提示词重新调用
raise error
# 使用示例
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.output_parsers import PydanticOutputParser
llm = ChatOpenAI(model="gpt-4")
prompt = ChatPromptTemplate.from_template("...")
chain = prompt | llm
robust_chain = RobustChain(chain)
result = robust_chain.invoke({"input": "测试输入"})4.3 性能优化
"""
性能优化策略
"""
import asyncio
from langchain.cache import InMemoryCache, SQLiteCache
from langchain.globals import set_llm_cache
from langchain_openai import ChatOpenAI
# 1. 启用缓存
set_llm_cache(InMemoryCache()) # 或 SQLiteCache("cache.db")
# 2. 异步调用
async def batch_invoke(chain, inputs):
"""批量异步调用"""
tasks = [chain.ainvoke(input_data) for input_data in inputs]
return await asyncio.gather(*tasks)
# 3. 流式输出
async def stream_response(chain, input_data):
"""流式输出"""
async for chunk in chain.astream(input_data):
yield chunk
# 4. 连接池优化
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4",
max_retries=3,
timeout=30,
request_timeout=60
)五、面试问答
Q1: LangChain 的核心组件有哪些？它们之间的关系是什么？
回答要点：
| 组件 | 作用 | 关系 |
|---|---|---|
| Prompt | 输入模板管理 | Chain 的输入层 |
| LLM | 模型调用 | Chain 的处理层 |
| Chain | 组件串联 | 连接所有组件 |
| Agent | 动态决策 | 使用 Tools 和 Memory |
| Tool | 外部能力 | Agent 的扩展 |
| Memory | 状态管理 | Agent 和 Chain 的上下文 |
关系图：
Prompt → LLM → Output Parser (基础链)
↓
+ Tools + Memory (Agent)
↓
最终输出Q2: LCEL（LangChain Expression Language）的优势是什么？
回答要点：
- 声明式语法：使用
|操作符直观连接组件 - 自动异步支持：无需额外代码即可使用异步
- 流式输出：内置流式处理能力
- 易于调试：每个步骤都可追踪
- 类型安全：更好的类型提示和验证
# 传统方式
chain = LLMChain(llm=llm, prompt=prompt)
# LCEL 方式
chain = prompt | llm | parser # 更简洁直观Q3: 如何选择合适的 Memory 类型？
回答要点：
| Memory 类型 | 适用场景 | Token 消耗 |
|---|---|---|
| BufferMemory | 短对话、需要完整历史 | 高 |
| WindowMemory | 中等对话、只关注近期 | 中 |
| SummaryMemory | 长对话、关注整体主题 | 低 |
| VectorMemory | 大规模历史、按相关性检索 | 中 |
选择建议：
- 对话轮数 < 10：BufferMemory
- 对话轮数 10-50：WindowMemory
- 对话轮数 > 50：SummaryMemory 或 VectorMemory
Q4: LangChain 如何实现工具调用？
回答要点：
- 工具定义：使用
@tool装饰器或继承BaseTool - 工具绑定：通过
llm.bind_tools(tools)或在 Agent 中配置 - 执行流程：LLM 生成工具调用 → 解析参数 → 执行工具 → 返回结果
- 错误处理：
handle_parsing_errors=True自动处理格式错误
Q5: LangChain 的局限性和改进方向？
回答要点：
局限性：
- 复杂工作流编排能力有限
- 状态管理不够灵活
- 调试和可观测性有待加强
- 学习曲线较陡峭
改进方向：
- LangGraph 提供更灵活的工作流
- LangSmith 提供可观测性平台
- 更多的内置工具和集成
- 简化的 API 设计
六、小结
LangChain 作为最流行的 Agent 开发框架，提供了构建 LLM 应用的完整工具链：
核心优势
- 生态丰富：支持多种 LLM、工具和向量存储
- 模块化设计：组件可灵活组合
- 生产就绪：缓存、重试、异步等企业级特性
关键要点
- Chain 是核心：理解 LCEL 语法是掌握 LangChain 的基础
- Agent 是目标：动态决策和工具调用是 Agent 的核心能力
- Memory 是关键：合适的记忆策略影响应用质量
- Tools 是扩展：丰富的工具生态扩展 Agent 能力
下一步学习
- 深入学习 LlamaIndex 的 RAG 能力
- 探索 LangGraph 的复杂工作流编排
- 实践生产级 Agent 应用的开发和部署