langchain 0.2
·
在实际应用中，我们经常需要：
- 🔄 将多个处理步骤串联起来
- 📊 让数据在不同组件之间流转
- 🎯 构建复杂的业务逻辑
举个例子：
假设你要开发一个"文章润色助手"：
- 第一步：分析文章的主题
- 第二步：根据主题生成改进建议
- 第三步：根据建议重写文章
问题是：如果每一步都单独调用模型，代码会非常冗长且难以维护：
# ❌ 传统方式：代码重复，难以维护
topic_response = llm.invoke("分析这篇文章的主题：" + article)
topic = topic_response.content
suggestions_response = llm.invoke(f"为{topic}主题的文章提供改进建议")
suggestions = suggestions_response.content
final_response = llm.invoke(f"根据建议{suggestions}重写文章")
result = final_response.content
LangChain的Chains就是为了解决这个问题！
它让你可以：
# ✅ Chains方式：简洁、可复用
chain = analyze_chain | suggest_chain | rewrite_chain
result = chain.invoke({"article": article})
Chain是将多个组件按顺序连接起来的处理流程。数据从第一个组件流向最后一个组件，每个组件的输出作为下一个组件的输入。
类比理解：
Chain就像工厂的生产线：
原材料 → [切割] → [打磨] → [上漆] → 成品
↓ ↓ ↓ ↓ ↓
输入 组件1 组件2 组件3 输出
┌──────────────────────────────────────┐
│ LangChain Chains 体系 │
├──────────────────────────────────────┤
│ │
│ 1. LLMChain（基础链） │
│ └─ 单个LLM + Prompt的组合 │
│ │
│ 2. SequentialChain（顺序链） │
│ ├─ SimpleSequentialChain │
│ │ └─ 简单的串行执行 │
│ └─ SequentialChain │
│ └─ 支持多输入输出的串行 │
│ │
│ 3. RouterChain（路由链） │
│ └─ 根据输入选择不同的处理分支 │
│ │
│ 4. LCEL（LangChain表达式语言） │
│ └─ 用 | 操作符连接组件（推荐） │
│ │
└──────────────────────────────────────┘
LLMChain - 最基础的链
【概念名称】：LLMChain
【来源】：langchain.chains.LLMChain
【作用】：将PromptTemplate和LLM组合在一起的最基本链
【核心组成】：
LLMChain = PromptTemplate + LLM + (可选)OutputParser
工作流程：
输入变量 → PromptTemplate格式化 → LLM处理 → 输出
【核心参数】：
LLMChain(
llm=ChatOpenAI(...), # 必需：要使用的语言模型
prompt=PromptTemplate(...), # 必需：提示词模板
output_parser=None, # 可选：输出解析器
verbose=False, # 可选：是否打印中间过程
memory=None # 可选：记忆模块（后续课程讲解）
)
参数详解：
-
llm（语言模型对象）
- 类型：BaseLanguageModel（如ChatOpenAI）
- 作用：执行实际的推理任务
- 示例：
ChatOpenAI(model_name="gpt-3.5-turbo")
-
prompt（提示词模板）
- 类型：PromptTemplate或ChatPromptTemplate
- 作用：定义如何格式化输入
- 必须包含：template和input_variables
-
verbose（详细输出模式）
- 类型：布尔值
- 作用：是否打印执行过程
- 值：
True：打印中间步骤（调试时有用）False：仅返回最终结果
- 示例输出：
> Entering new LLMChain chain...
Prompt after formatting:
请翻译：Hello
> Finished chain.
- output_parser（输出解析器）
- 类型：BaseOutputParser
- 作用：将模型输出转换为特定格式
- 示例：CommaSeparatedListOutputParser()
SequentialChain - 顺序链
【概念名称】：SequentialChain
【作用】：按顺序执行多个链，每个链的输出可以传递给下一个链
【与SimpleSequentialChain的区别】：
SimpleSequentialChain：
├─ 限制：每个链只能有1个输入和1个输出
├─ 优点：简单直观
└─ 适用场景：简单的串行任务
链1 → [输出] → 链2 → [输出] → 链3
SequentialChain：
├─ 支持：每个链可以有多个输入和多个输出
├─ 优点：灵活强大
└─ 适用场景：复杂的业务逻辑
链1 → [输出A, 输出B] → 链2(使用A) → [输出C] → 链3(使用B和C)
【核心参数】：
SequentialChain(
chains=[chain1, chain2, chain3], # 要执行的链列表
input_variables=["input"], # 初始输入变量名
output_variables=["output"], # 最终输出变量名
verbose=False # 是否打印过程
)
RouterChain - 路由链
【概念名称】：RouterChain
【作用】：根据输入内容动态选择要执行的链
【工作原理】：
输入
↓
[路由决策器]
/ | \
/ | \
链A（技术） 链B（法律） 链C（医疗）
\ | /
\ | /
输出
【使用场景】：
- 多领域问答系统（根据问题类型选择专家模型）
- 智能客服分流（根据问题分配给不同部门）
- 多语言处理（根据语言选择不同的处理链）
LCEL - LangChain表达式语言（推荐）
【概念名称】：LCEL (LangChain Expression Language)
【作用】：使用 | 操作符连接组件的现代化语法
【为什么推荐LCEL？】
-
代码更简洁：
# ❌ 传统LLMChain方式 chain = LLMChain(llm=llm, prompt=prompt) result = chain.invoke({"input": "..."}) # ✅ LCEL方式 chain = prompt | llm | output_parser result = chain.invoke({"input": "..."}) -
更易组合：
# 轻松添加新组件 chain = prompt | llm | output_parser | custom_processor -
支持流式输出：
# 实时获取模型输出（类似ChatGPT打字效果） for chunk in chain.stream({"input": "..."}): print(chunk, end="", flush=True) -
更好的类型检查：
- LCEL会自动检查组件间的输入输出是否匹配
Memory
Memory就是为了解决无记忆！
它让模型能够：
- 💬 记住之前的对话
- 🧠 理解上下文关系
- 🔄 进行连贯的多轮对话
Memory的工作原理：
用户输入 → Memory读取历史 → 组合成完整Prompt → LLM处理 → Memory保存新对话 → 输出
↑ ↓
└──────────────────── 存储空间 ←──────────────────────┘
Memory模块的核心组成：
BaseMemory（抽象基类）
├─ memory_variables: 定义传递给Prompt的变量名
├─ load_memory_variables(): 读取记忆数据
├─ save_context(): 保存对话到记忆
└─ clear(): 清空记忆
BaseMemory接口
【概念名称】：BaseMemory
【来源】：langchain.schema.BaseMemory
【作用】：所有Memory类的抽象基类，定义了Memory必须实现的接口
【核心方法】：
-
memory_variables（属性）
@property @abstractmethod def memory_variables(self) -> List[str]: """返回这个Memory类会添加到链输入的变量名列表"""- 作用：定义Memory会向Prompt注入哪些变量
- 示例：返回
["history"]表示会注入{history}变量
-
load_memory_variables（方法）
@abstractmethod def load_memory_variables(self, inputs: Dict[str, Any]) -> Dict[str, Any]: """根据输入加载相关的记忆数据"""- 作用：读取存储的记忆，返回要注入Prompt的内容
- 参数：inputs - 当前输入（可用于过滤相关记忆）
- 返回：字典，键为memory_variables中定义的变量名
-
save_context（方法）
@abstractmethod def save_context(self, inputs: Dict[str, Any], outputs: Dict[str, str]) -> None: """保存这次对话到记忆中"""- 作用：将输入和输出保存到记忆存储
- 参数：
- inputs：用户输入
- outputs：模型输出
-
clear（方法）
@abstractmethod def clear(self) -> None: """清空所有记忆"""
实现自定义Memory的步骤
步骤1：继承BaseMemory
↓
步骤2：定义存储变量（如dict、list）
↓
步骤3：实现memory_variables属性
↓
步骤4：实现load_memory_variables()
↓
步骤5：实现save_context()
↓
步骤6：实现clear()
LangChain内置Memory类型
内置Memory家族
├─ ConversationBufferMemory（最基础）
│ └─ 完整保存所有对话历史
├─ ConversationBufferWindowMemory（有限历史）
│ └─ 只保存最近K轮对话
├─ ConversationEntityMemory（实体记忆）
│ └─ 基于实体提取的智能记忆
├─ ConversationSummaryMemory（摘要记忆）
│ └─ 用LLM压缩历史为摘要
└─ ChatMessageHistory（消息历史）
└─ 底层消息存储类
ConversationBufferMemory
【概念】：最简单的Memory，保存所有对话历史
【原理】：
第1轮：用户: Hello
AI: Hi
↓
Memory: "Human: Hello\nAI: Hi"
第2轮：用户: How are you?
AI: I'm fine
↓
Memory: "Human: Hello\nAI: Hi\nHuman: How are you?\nAI: I'm fine"
【代码示例】：
from langchain.memory import ConversationBufferMemory
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
# 【创建Memory】
memory = ConversationBufferMemory()
# 【手动测试Memory】
# 保存对话
memory.save_context(
{"input": "你好"}, # 用户输入
{"output": "你好，有什么可以帮你的？"} # AI输出
)
# 加载历史
history = memory.load_memory_variables({})
print(history)
# 输出：{'history': 'Human: 你好\nAI: 你好，有什么可以帮你的？'}
# 【集成到Chain】
llm = ChatOpenAI(model_name="gpt-3.5-turbo")
prompt = PromptTemplate(
input_variables=["history", "input"],
template="""以下是对话历史：
{history}
人类: {input}
AI:"""
)
chain = LLMChain(
llm=llm,
prompt=prompt,
memory=ConversationBufferMemory(memory_key="history"), # 指定memory_key
verbose=True
)
# 对话
chain.invoke({"input": "我叫张三"})
chain.invoke({"input": "我叫什么名字？"}) # AI会记得"张三"
【参数详解】：
memory_key：在Prompt中的变量名（默认"history"）return_messages：是否返回消息对象（默认False）# return_messages=False（默认） {'history': 'Human: 你好\nAI: 你好'} # return_messages=True {'history': [HumanMessage(content='你好'), AIMessage(content='你好')]}
【优点】：
✅ 简单易用
✅ 保留完整上下文
【缺点】：
❌ 历史过长会超出Token限制
❌ 成本随对话增加而增加
ConversationBufferWindowMemory
【概念】：只保存最近K轮对话的Memory
【原理】：
设置k=2（只保存最近2轮）
第1轮：用户: A, AI: B
Memory: A, B
第2轮：用户: C, AI: D
Memory: A, B, C, D
第3轮：用户: E, AI: F
Memory: C, D, E, F ← 删除了A, B（最旧的）
【代码示例】：
from langchain.memory import ConversationBufferWindowMemory
# 【创建WindowMemory】
memory = ConversationBufferWindowMemory(
k=2 # 只保留最近2轮对话
)
# 【测试】
memory.save_context({"input": "第1轮"}, {"output": "回复1"})
memory.save_context({"input": "第2轮"}, {"output": "回复2"})
memory.save_context({"input": "第3轮"}, {"output": "回复3"})
history = memory.load_memory_variables({})
print(history)
# 输出：只包含第2轮和第3轮（第1轮被丢弃）
# 【集成到Chain】
chain = LLMChain(
llm=llm,
prompt=prompt,
memory=ConversationBufferWindowMemory(k=3, memory_key="history"),
verbose=True
)
【适用场景】：
- ✅ 长期对话（避免超Token）
- ✅ 注重最近上下文的应用
- ✅ 需要控制成本
【权衡】：
- 设置太小（k=1-2）：可能丢失重要信息
- 设置太大（k=10+）：与BufferMemory差别不大
ConversationEntityMemory
【概念】：基于实体提取的智能Memory（使用LLM识别实体）
【原理】：
输入："马云在杭州创办了阿里巴巴"
↓ LLM提取实体
实体：
- 马云: "在杭州创办了阿里巴巴"
- 杭州: "马云在杭州创办了阿里巴巴"
- 阿里巴巴: "马云在杭州创办"
下次提到"马云"时，自动加载相关信息
【代码示例】：
from langchain.memory import ConversationEntityMemory
# 【创建EntityMemory】
# 需要LLM来提取实体
memory = ConversationEntityMemory(llm=llm)
# 【测试】
memory.save_context(
{"input": "孙悟空和猪八戒正在做西天取经项目"},
{"output": "听起来很棒！"}
)
# 查询时，如果提到实体，会加载相关信息
result = memory.load_memory_variables({"input": "谁是孙悟空？"})
print(result)
# 输出包含：孙悟空相关的上下文信息
# 【集成到Chain】
from langchain.prompts import PromptTemplate
template = """相关实体信息：
{entities}
当前对话：
{history}
人类: {input}
AI:"""
prompt = PromptTemplate(
input_variables=["entities", "history", "input"],
template=template
)
chain = LLMChain(
llm=llm,
prompt=prompt,
memory=ConversationEntityMemory(llm=llm),
verbose=True
)
【优点】：
✅ 智能定位相关信息
✅ 适合长期对话
✅ 避免无关历史干扰
【缺点】：
❌ 需要额外LLM调用（成本更高）
❌ 实体提取可能不准确
ConversationSummaryMemory
【概念】：使用LLM将历史对话压缩为摘要
【原理】：
历史对话（500 tokens）：
"Human: 我叫张三
AI: 你好张三
Human: 我在北京工作
AI: 北京是个好地方
..."
↓ LLM压缩
摘要（50 tokens）：
"人类介绍自己叫张三，在北京工作。"
【代码示例】：
from langchain.memory import ConversationSummaryMemory
# 【创建SummaryMemory】
# 需要LLM来生成摘要
memory = ConversationSummaryMemory(llm=llm)
# 【测试】
memory.save_context(
{"input": "我是孙悟空，正在西天取经"},
{"output": "祝你旅途顺利"}
)
memory.save_context(
{"input": "我已经经历了九九八十一难"},
{"output": "真不容易！"}
)
# 查看摘要
summary = memory.load_memory_variables({})
print(summary)
# 输出：压缩后的摘要，而不是完整对话
# 【自定义摘要Prompt】
from langchain.prompts import PromptTemplate
# 定义如何生成摘要
SUMMARY_PROMPT = PromptTemplate(
input_variables=["summary", "new_lines"],
template="""当前摘要：
{summary}
新对话：
{new_lines}
生成新摘要（简洁、准确）："""
)
memory = ConversationSummaryMemory(
llm=llm,
prompt=SUMMARY_PROMPT
)
# 【集成到Chain】
chain = LLMChain(
llm=llm,
prompt=prompt,
memory=ConversationSummaryMemory(llm=llm),
verbose=True
)
【优点】：
✅ 大幅节省Token
✅ 适合超长对话
✅ 保留关键信息
【缺点】：
❌ 每次都调用LLM（成本）
❌ 可能丢失细节
❌ 摘要质量依赖LLM能力
ChatMessageHistory
【概念】：底层消息存储类，其他Memory的基础
【代码示例】：
from langchain.memory import ChatMessageHistory
# 【创建MessageHistory】
history = ChatMessageHistory()
# 【添加消息】
history.add_user_message("你好") # 用户消息
history.add_ai_message("你好，有什么可以帮你的？") # AI消息
# 【查看消息】
print(history.messages)
# 输出：[HumanMessage(content='你好'), AIMessage(content='你好，有什么可以帮你的？')]
# 【单独使用】
# 传给LLM
llm = ChatOpenAI()
response = llm.invoke(history.messages)
print(response.content)
【使用场景】：
- 需要完全自定义Memory逻辑
- 与其他Memory类组合使用
| Memory类型 | 存储内容 | Token消耗 | 适用场景 | 额外成本 |
|---|---|---|---|---|
| ConversationBuffer | 完整历史 | 随对话增长 | 短对话 | 无 |
| ConversationBufferWindow | 最近K轮 | 固定上限 | 长对话 | 无 |
| ConversationEntity | 实体信息 | 中等 | 需要记住特定信息 | LLM调用 |
| ConversationSummary | 摘要 | 最少 | 超长对话 | LLM调用 |
选择建议：
对话轮次 < 10轮：
└─ ConversationBufferMemory（简单直接）
10轮 < 对话轮次 < 50轮：
└─ ConversationBufferWindowMemory（平衡性能）
对话轮次 > 50轮：
└─ ConversationSummaryMemory（节省Token）
需要精确记住特定信息（如人名、地点）：
└─ ConversationEntityMemory
Model I/O
直接调用OpenAI API虽然可行，但存在以下挑战：
- 提示词难以复用和管理
- 输出格式不统一，难以解析
- 需要手动处理API调用的各种细节
LangChain的Model I/O模块就是为了解决这些问题而设计的！
🏗️ 它把常用的操作封装成标准化的组件，让你更高效地构建AI应用
LangChain的核心价值：
原始方式：
你 → 写提示词 → 调用API → 手动解析 → 得到结果
LangChain方式：
你 → 使用PromptTemplate → 调用LLM → 使用OutputParser → 得到结构化结果
↓ ↓ ↓
可复用 统一接口 自动解析
核心组件：
┌─────────────────────────────────────┐
│ LangChain Model I/O │
├─────────────────────────────────────┤
│ 1. Prompts（提示词管理） │
│ └─ PromptTemplate │
│ └─ ChatPromptTemplate │
│ │
│ 2. Language Models（语言模型） │
│ └─ ChatOpenAI │
│ └─ OpenAI │
│ │
│ 3. Output Parsers（输出解析器） │
│ └─ StrOutputParser │
│ └─ PydanticOutputParser │
│ └─ CommaSeparatedListParser │
└─────────────────────────────────────┘
PromptTemplate核心概念
【概念名称】：PromptTemplate
【来源】：langchain.prompts 模块
【作用】：创建可复用的提示词模板，支持变量替换
【核心参数】：
template（str）：包含占位符的模板字符串，使用{变量名}标记占位符input_variables（List[str]）：模板中使用的变量名列表template_format（str）：模板格式，默认为"f-string"（Python f-string格式）
【工作原理】：
模板定义：
"你是{role}，请{task}"
↓
变量替换：
role="翻译专家", task="翻译以下文本"
↓
最终提示词：
"你是翻译专家，请翻译以下文本"
ChatPromptTemplate - 对话式提示词
【概念名称】：ChatPromptTemplate
【来源】：langchain.prompts 模块
【作用】：专门为聊天模型（如GPT-3.5/4）设计的提示词模板
【与PromptTemplate的区别】：
PromptTemplate：
└─ 适用于：补全模型（Completion Models）
└─ 格式：单一字符串
└─ 示例："请回答：{question}"
ChatPromptTemplate：
└─ 适用于：聊天模型（Chat Models）
└─ 格式：消息列表（system, human, ai）
└─ 示例：[
{"role": "system", "content": "你是AI助手"},
{"role": "human", "content": "{question}"}
]
【消息角色说明】：
system：系统消息，定义AI的行为和角色human（或user）：用户消息，来自人类的输入ai（或assistant）：AI消息，模型的回复
ChatOpenAI核心类
【概念名称】：ChatOpenAI
【来源】：langchain_openai 包
【作用】：LangChain对OpenAI聊天模型的封装类
【核心参数详解】：
ChatOpenAI(
model_name="gpt-3.5-turbo", # 使用的模型名称
temperature=0.7, # 创造性参数
max_tokens=None, # 最大生成token数
openai_api_key=None, # API密钥
openai_api_base=None, # API基础URL
streaming=False, # 是否流式输出
verbose=False # 是否打印详细信息
)
参数详细说明：
-
model_name（字符串）
- 作用：指定使用的OpenAI模型
- 常用值：
"gpt-3.5-turbo"：性价比高，适合日常对话"gpt-4"：能力更强，推理能力好"gpt-4-turbo"：速度更快的GPT-4
- 默认值：
"gpt-3.5-turbo"
-
temperature（浮点数，范围0-2）
- 作用：控制输出的随机性和创造性
- 取值说明：
0.0：确定性输出，每次结果几乎相同0.5-0.7：平衡的创造性（推荐）1.0-2.0：高度创造性，输出更多样化
- 使用建议：
- 事实性任务（如翻译）：使用低温度（0-0.3）
- 创意任务（如写作）：使用高温度（0.7-1.0）
- 默认值：
0.7
-
max_tokens（整数或None）
- 作用：限制模型生成的最大token数量
- 说明：1个token ≈ 4个英文字符 或 1-2个中文字符
- 示例：
max_tokens=100：约生成100-200个中文字max_tokens=None：使用模型默认限制
- 注意：设置过小可能导致回答不完整
-
streaming（布尔值）
- 作用：是否启用流式输出（像ChatGPT一样逐字显示）
- 值：
True：启用流式输出False：等待完整响应后一次性返回
- 使用场景：实时聊天应用建议开启
模型调用方式
LangChain的ChatOpenAI提供两种调用方式：
# 方式1：invoke() - 同步调用（推荐用于简单场景）
response = llm.invoke("你好")
# 方式2：ainvoke() - 异步调用（推荐用于高并发场景）
response = await llm.ainvoke("你好")
Output Parsers - 输出解析器
为什么需要输出解析器？
问题场景：
模型的原始输出是字符串，但程序往往需要结构化数据。
# 模型输出（字符串）
"推荐的水果有：苹果, 香蕉, 橙子"
# 程序需要的格式（列表）
['苹果', '香蕉', '橙子']
OutputParser就是做这个转换的！
常用输出解析器
【概念名称】：StrOutputParser
【作用】：将模型输出转换为字符串（最简单的解析器）
【使用场景】：只需要文本结果，不需要特殊处理
【概念名称】：CommaSeparatedListOutputParser
【作用】：将逗号分隔的字符串解析为列表
【输入示例】："苹果, 香蕉, 橙子"
【输出结果】：['苹果', '香蕉', '橙子']
【概念名称】：PydanticOutputParser
【作用】：将输出解析为Pydantic模型（结构化对象）
【使用场景】：需要复杂的结构化数据（如JSON对象）
Retrieval
在实际的AI应用开发中，我们经常遇到这样的需求：
- 📚 让大模型回答基于企业私有知识库的问题
- 💼 构建能够理解和检索大量文档的智能系统
- 🔍 实现精准的信息检索和问答系统
问题是：大模型虽然强大，但它存在几个关键限制：
- 知识截止日期：模型只知道训练时的数据，无法获取最新信息
- 无法访问私域数据：企业内部文档、数据库等私有信息模型无法直接访问
- 上下文窗口限制：即使是最强的模型也有Token数量限制，无法一次性处理海量文档
- 幻觉问题：模型可能生成看似合理但实际错误的信息
RAG（检索增强生成）就是为了解决这些问题而设计的！
完整RAG流程图：
1. 文档加载
TextLoader → Document对象
↓
2. 文本分割
CharacterTextSplitter → 多个小块
↓
3. 向量化
OpenAIEmbeddings → 每块转为向量
↓
4. 存储
Chroma → 向量数据库
↓
5. 检索
用户问题 → 向量化 → 相似度搜索 → Top-K文档
↓
6. 生成
检索结果 + 问题 → LLM → 答案
Retrieval 体系：
┌─────────────────────────────────────────┐
│ LangChain Retrieval 体系 │
├─────────────────────────────────────────┤
│ 1. Document Loaders（文档加载器） │
│ └─ TextLoader, PDFLoader, JSONLoader│
│ │
│ 2. Text Splitters（文本分割器） │
│ └─ RecursiveCharacterTextSplitter │
│ └─ TokenTextSplitter │
│ │
│ 3. Text Embeddings（文本向量化） │
│ └─ OpenAIEmbeddings │
│ │
│ 4. Vector Stores（向量存储） │
│ └─ Chroma, FAISS │
│ │
│ 5. Retrievers（检索器） │
│ └─ VectorStoreRetriever │
└─────────────────────────────────────────┘
✅ LangChain Retrieval模块的五大组件：
- Document Loaders：统一加载各种格式文档
- Text Splitters：智能分割文本，兼顾语义和长度
- Text Embeddings：将文本转为向量表示
- Vector Stores：高效存储和检索向量
- Retrievers：统一的检索接口
Document Loaders - 文档加载器
为什么需要Document Loaders？
问题场景：
企业的知识库通常包含多种格式的文档：
- 📄 PDF技术文档
- 📝 TXT文本文件
- 📊 JSON数据文件
- 🌐 网页内容
- 💾 数据库记录
如果没有统一的加载机制：
# ❌ 每种格式需要不同的处理方式
pdf_content = extract_pdf(file) # PDF专用方法
txt_content = read_text(file) # TXT专用方法
json_content = parse_json(file) # JSON专用方法
# 代码重复、难以维护
Document Loaders的解决方案：
# ✅ 统一的加载接口
from langchain.document_loaders import PDFLoader, TextLoader, JSONLoader
# 所有Loader都使用相同的.load()方法
pdf_docs = PDFLoader("file.pdf").load()
txt_docs = TextLoader("file.txt").load()
json_docs = JSONLoader("file.json").load()
# 返回统一的Document对象
Document对象详解
【概念名称】：Document
【来源】：langchain_core.documents.base
【作用】：LangChain中统一的文档数据结构
【核心属性】：
class Document:
page_content: str # 文档的实际内容
metadata: dict # 文档的元数据信息
属性详解：
-
page_content（字符串）
- 作用：存储文档的实际文本内容
- 示例：
doc.page_content = "LangChain是一个用于开发大模型应用的框架..." -
metadata（字典）
- 作用：存储文档的附加信息
- 常见字段：
source：文档来源路径page：页码（PDF文档）author：作者created_at：创建时间
- 示例：
doc.metadata = { "source": "./data/langchain.txt", "author": "Claude", "page": 1 }
BaseLoader基类
【概念名称】：BaseLoader
【作用】：所有文档加载器的抽象基类，定义统一接口
【核心方法】：
class BaseLoader(ABC):
@abstractmethod
def load(self) -> List[Document]:
"""加载文档并返回Document对象列表"""
pass
def load_and_split(
self,
text_splitter: Optional[TextSplitter] = None
) -> List[Document]:
"""加载文档并自动分割成块"""
pass
方法说明：
-
load()
- 作用：从数据源加载文档
- 返回：
List[Document] - 必须实现：是（抽象方法）
-
load_and_split()
- 作用：加载文档并自动使用TextSplitter分割
- 参数：
text_splitter：可选，默认使用RecursiveCharacterTextSplitter
- 返回：分割后的
List[Document]
Text Splitters - 文本分割器
为什么需要Text Splitters？
问题1：相关性问题
场景：一个100页的PDF技术文档
问题：用户问"什么是LangChain？"
答案在第5页
如果不分割：
❌ 整个100页都送入模型
❌ 大量无关信息干扰
❌ 模型可能找不到重点
如果分割：
✅ 只检索第5页的相关段落
✅ 信息精准、干扰少
✅ 模型回答更准确
问题2：Token限制
GPT-4模型限制：最多8192个Token
一个100页PDF ≈ 50,000 Token
↓
❌ 无法一次性输入
分割后每块 ≈ 500 Token
↓
✅ 可以检索并输入相关块
Chunking策略对比
| 策略 | 原理 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 按句子分割 | 使用标点符号 | 语义完整 | 长度不均 | 结构化文本 |
| 固定字符数 | 每N个字符切一次 | 长度统一 | 可能破坏语义 | 日志文件 |
| 固定字符+重叠 | 字符数+窗口重叠 | 保留上下文 | 有冗余 | 一般文本 |
| 递归分割 | 按层级标记切分 | 兼顾语义和长度 | 稍复杂 | 推荐方案 |
| 语义分割 | 基于语义理解 | 语义最完整 | 效率低 | 高精度需求 |
RecursiveCharacterTextSplitter详解
【概念名称】：RecursiveCharacterTextSplitter
【来源】：langchain_text_splitters
【作用】：LangChain默认的文本分割器，递归按分隔符分割
【核心参数】：
RecursiveCharacterTextSplitter(
separators=["\n\n", "\n", " ", ""], # 分隔符列表（优先级从高到低）
chunk_size=1000, # 每块的最大字符数
chunk_overlap=200, # 块之间的重叠字符数
length_function=len, # 计算长度的函数
add_start_index=True # 是否添加起始索引到metadata
)
参数详解：
-
separators（分隔符列表）
- 类型：
List[str] - 作用：定义分割文本的标记，按优先级使用
- 默认值：
["\n\n", "\n", " ", ""] - 工作原理：
先尝试用"\n\n"（段落）分割 ↓ 如果块还是太大 再用"\n"（行）分割 ↓ 如果还是太大 再用" "（空格）分割 ↓ 最后 按字符""分割
- 类型：
-
chunk_size（块大小）
- 类型：
int - 作用：每个文本块的最大长度
- 建议值：
- 短文档：500-1000
- 长文档：1000-2000
- 技术文档：1500-2500
- 注意：这是触发分割的阈值，实际块可能略大
- 类型：
-
chunk_overlap（重叠长度）
- 类型：
int - 作用：相邻块之间重叠的字符数
- 为什么需要重叠：
块1：...重要信息在这里→| 块2： |←这里继续... ↑ 重叠区域（保留上下文） - 建议值：chunk_size的10-20%
- 类型：
-
length_function（长度函数）
- 类型：
Callable[[str], int] - 作用：如何计算文本长度
- 选项：
len：按字符数（默认）- 自定义Token计数函数
- 类型：
-
add_start_index（添加起始索引）
- 类型：
bool - 作用：是否在metadata中记录块在原文的起始位置
- 用途：可追溯信息来源的精确位置
- 类型：
按Token分割
【概念名称】：Token
【定义】：模型处理文本的最小单位，约等于一个词或子词
Token与字符的关系：
英文：1 Token ≈ 4 个字符
"Hello World" ≈ 2 Tokens
中文：1 个汉字 ≈ 1-2 Tokens
"你好世界" ≈ 4 Tokens
为什么按Token分割：
GPT-4的限制：8192 Tokens（不是字符！）
如果按字符分割：
chunk_size=1000字符 ≈ 250 Tokens（英文）
❌ 可能无法充分利用模型容量
如果按Token分割：
chunk_size=1000 Tokens
✅ 精确控制输入量
✅ 避免超出限制
✅ 成本计算准确
Text Embeddings - 文本向量化
什么是Embedding？
【概念名称】：Embedding（嵌入向量）
【作用】：将文本转换为数字向量，使计算机能够理解文本语义
类比理解：
文本 = 人类语言
向量 = 计算机语言
"苹果" → [0.2, -0.5, 0.8, ...] # 1536维向量
"香蕉" → [0.3, -0.4, 0.7, ...] # 相似的向量
"汽车" → [-0.8, 0.9, -0.1, ...] # 不相似的向量
为什么需要Embedding：
问题：如何判断两段文本是否相关？
字符匹配：
"LangChain是一个框架"
"框架是什么？"
❌ 几乎没有相同字符，但语义相关
向量相似度：
向量1 = [0.1, 0.9, ...]
向量2 = [0.2, 0.8, ...]
✅ 余弦相似度 = 0.95（高度相关）
OpenAIEmbeddings
【概念名称】：OpenAIEmbeddings
【来源】：langchain_openai
【作用】：使用OpenAI的embedding模型将文本转为向量
【核心参数】：
OpenAIEmbeddings(
model="text-embedding-ada-002", # 模型名称
openai_api_key="your-key", # API密钥
openai_api_base="https://..." # API基础URL
)
主要方法：
-
embed_documents(texts: List[str])
- 作用：批量向量化多个文档
- 参数：文本列表
- 返回：向量列表
- 用途：处理知识库文档
-
embed_query(text: str)
- 作用：向量化单个查询
- 参数：查询文本
- 返回：单个向量
- 用途：处理用户问题
Vector Stores - 向量存储
为什么需要Vector Store？
问题场景：
知识库有10,000个文档块
用户问："LangChain是什么？"
如果没有向量存储：
❌ 需要遍历10,000个文档
❌ 逐个计算相似度
❌ 速度极慢
有向量存储：
✅ 预先计算并索引所有向量
✅ 快速检索最相似的Top-K
✅ 毫秒级响应
Chroma向量数据库
【概念名称】：Chroma
【来源】：langchain_chroma
【作用】：轻量级的向量数据库，适合开发和小规模应用
【核心方法】：
# 创建向量存储
db = Chroma.from_documents(
documents=docs, # Document对象列表
embedding=embeddings # Embedding模型
)
# 相似度搜索
results = db.similarity_search(
query="查询文本", # 查询内容
k=4 # 返回Top-K个结果
)
Retrievers - 检索器
【概念名称】：Retriever
【作用】：统一的检索接口，从向量存储中检索相关文档
核心方法：
retriever = db.as_retriever()
docs = retriever.invoke("查询文本")
实战案例1：加载不同格式的文档
# ========================================
# 示例：使用Document Loaders加载多种格式文档
# ========================================
# 【导入必要模块】
from langchain.document_loaders import TextLoader
from langchain_community.document_loaders import PyPDFLoader, JSONLoader
# ==================== 加载TXT文档 ====================
# 【创建TextLoader】
# 参数：
# - file_path：文件路径
# - encoding：文件编码（中文文件必须指定utf-8）
txt_loader = TextLoader(
'./data/langchain.txt',
encoding="utf-8"
)
# 【调用load方法】
# 作用：读取文件并转换为Document对象
txt_docs = txt_loader.load()
# 【查看结果】
print("=== TXT文档加载结果 ===")
print(f"文档数量：{len(txt_docs)}") # 输出：1
print(f"文档类型：{type(txt_docs[0])}") # 输出：Document
print(f"文档内容（前30字符）：{txt_docs[0].page_content[:30]}")
print(f"文档元数据：{txt_docs[0].metadata}")
# 输出：{'source': './data/langchain.txt'}
# ==================== 加载PDF文档 ====================
# 【创建PyPDFLoader】
# 注意：PyPDFLoader会自动将PDF的每一页作为一个Document
pdf_loader = PyPDFLoader("./data/本地知识库.pdf")
# 【load方法】：加载所有页
pdf_docs = pdf_loader.load()
# 【load_and_split方法】：加载并分割
# 作用：加载PDF的同时进行文本分割
pdf_docs_split = pdf_loader.load_and_split()
# 【查看结果】
print("\n=== PDF文档加载结果 ===")
print(f"总页数：{len(pdf_docs)}")
print(f"第一页内容（前50字符）：{pdf_docs[0].page_content[:50]}")
print(f"第一页元数据：{pdf_docs[0].metadata}")
# 输出：{'source': './data/本地知识库.pdf', 'page': 0}
# ==================== 加载JSON文档 ====================
# 【安装jq库】
# 在命令行执行：pip install jq
# 【创建JSONLoader】
# 参数：
# - file_path：JSON文件路径
# - jq_schema：jq语法表达式，用于提取JSON中的特定字段
# - text_content：是否将结果作为文本内容
# 示例：提取JSON中的commentary字段
json_loader = JSONLoader(
file_path='./data/test.json',
jq_schema='.[].commentary', # jq语法：提取数组中每个元素的commentary字段
text_content=False
)
# 【加载数据】
json_docs = json_loader.load()
# 【查看结果】
print("\n=== JSON文档加载结果 ===")
print(f"文档数量：{len(json_docs)}")
print(f"第一个文档内容（前100字符）：{json_docs[0].page_content[:100]}")
关键知识点：
-
统一的接口设计：
# 所有Loader都有相同的方法 docs = loader.load() # 加载文档 docs = loader.load_and_split() # 加载并分割 -
Document对象的访问：
doc = docs[0] content = doc.page_content # 访问内容 meta = doc.metadata # 访问元数据 -
jq语法基础：
# JSON结构：[{"text": "..."}, {"text": "..."}] jq_schema = ".[].text" # 提取所有text字段 # JSON结构：{"key": [{"text": "..."}, ...]} jq_schema = ".key[].text" # 提取key下所有text字段
MCP
实际应用中常遇到这样的需求:
- 🔄 需要同时使用多个不同的服务(天气、数据库、Python执行等)
- 🌐 每个服务可能运行在不同的进程中
- 🔗 需要统一管理和调用这些服务
问题场景:
用户:"请查询北京天气,并将结果存入数据库"
需要调用:
❌ 单一服务无法完成
✅ 需要天气服务 + 数据库服务 + Python服务配合
MCP多服务器架构就是为了解决这个问题而设计的!
什么是MCP多服务器架构?
核心定义:
MCP(Model Context Protocol)多服务器架构是一种能够同时管理多个独立MCP服务器的客户端架构。每个服务器提供特定领域的工具,客户端负责统一调度和协调。
工作流程:
【MultiServerMCPClient】
↓
┌────────────────┼────────────────┐
↓ ↓ ↓
[SQL Server] [Python Server] [Weather Server]
↓ ↓ ↓
sql_inter() python_inter() query_weather()
↓ ↓ ↓
MySQL数据库 Python运行时 天气API
类比理解:
MCP多服务器架构 就像一个项目经理:
单服务器 = 一个专家独立工作
↓
问题:能力有限,无法处理跨领域任务
多服务器 = 项目经理协调多个专家
↓
1. 识别任务需要哪些专家(工具选择)
2. 分配任务给不同专家(工具调用)
3. 汇总结果(结果整合)
核心组件
MCP多服务器架构组成
├─ MultiServerMCPClient (客户端)
│ ├─ 管理多个MCP服务器连接
│ ├─ 统一工具列表
│ ├─ Function Calling集成
│ └─ 异步通信管理
│
├─ MCP服务器 (Server)
│ ├─ FastMCP框架
│ ├─ @mcp.tool()装饰器
│ └─ stdio通信
│
└─ OpenAI Function Calling
└─ 智能工具选择和调用
FastMCP框架
【概念名称】:FastMCP
【来源】:mcp.server.fastmcp
【作用】:快速构建MCP服务器的框架
【核心特性】:
from mcp.server.fastmcp import FastMCP
# 【创建MCP服务器】
mcp = FastMCP("ServerName") # ServerName:服务器标识
# 【定义工具】使用装饰器
@mcp.tool()
async def tool_name(param1, param2):
"""
工具描述(会被LLM看到,用于理解工具功能)
:param param1: 参数1说明
:param param2: 参数2说明
:return: 返回值说明
"""
# 工具实现逻辑
return result
# 【启动服务器】
if __name__ == "__main__":
mcp.run(transport='stdio') # 使用标准输入输出通信
StdioServerParameters
【概念名称】:StdioServerParameters
【作用】:配置MCP服务器子进程的启动参数
【核心参数】:
from mcp import StdioServerParameters
server_params = StdioServerParameters(
command="python", # 执行命令(python或node)
args=["server_script.py"], # 脚本路径
env=None # 环境变量(None表示继承父进程)
)
参数详解:
-
command(字符串)
- 作用:指定服务器脚本的解释器
- 选项:
"python":Python脚本"node":JavaScript脚本
- 示例:
"python"
-
args(列表)
- 作用:传递给命令的参数
- 通常是脚本文件路径
- 示例:
["SQL_server.py"]
-
env(字典或None)
- 作用:设置环境变量
- None:继承父进程环境
- 示例:
{"API_KEY": "xxx"}
ClientSession
【概念名称】:ClientSession
【作用】:MCP客户端与服务器的通信会话
【核心方法】:
from mcp import ClientSession
# 1. 初始化会话
await session.initialize()
# 2. 列出服务器提供的工具
response = await session.list_tools()
tools = response.tools # 工具列表
# 3. 调用工具
result = await session.call_tool(
name="tool_name", # 工具名称
arguments={"param": "value"} # 参数字典
)
# 4. 获取结果
content = result.content
Function Calling格式转换
【问题】:Claude和OpenAI的Function Calling格式不同
Claude格式:
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询天气",
"input_schema": { # ⚠️ Claude使用input_schema
"type": "object",
"properties": {...},
"required": [...]
}
}
}
OpenAI格式:
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询天气",
"parameters": { # ⚠️ OpenAI使用parameters
"type": "object",
"properties": {...},
"required": [...]
}
}
}
转换逻辑:
# input_schema → parameters
# 其他字段保持不变
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
30
0- 0
已为社区贡献4条内容
所有评论(0)