LangChain教程-2、Langchian基础

凶契帽 · 7 小时前

LangChain 1.0 实战教程 · 10 个 Demo

基于 LangChain ^1.0 版本（1.x 主线能力）
每个 Demo 含完整代码 + 每行逐词解析 + 作用说明
前置要求：Python 3.11+、uv、API Key（OpenAI / 硅基流动等兼容接口）

开场：先讲清 LangChain 的 3W1H

1) Why：为什么需要 LangChain？

直接调用模型 API 只能“单次问答”，但真实业务需要的是“系统能力”：提示词管理、工具调用、检索（RAG）、多步编排、可观测性、部署。LangChain 的价值，就是把这些“应用层能力”标准化，减少你重复造轮子。
2) What：LangChain 是什么？

LangChain 是一个 LLM 应用框架，核心是“可组合组件”：

模型层：ChatOpenAI 等模型适配器
编排层：LCEL（| 管道）与 Runnable 统一接口
数据层：Loader / Splitter / Vector Store / Retriever
能力层：Tool / Agent / Memory / Parser
工程层：Callback / LangSmith / LangServe

3) Who：谁应该用？

个人开发者：想快速做出可运行的 AI 功能
后端工程师：要把“提示词脚本”升级为可维护服务
AI 应用团队：需要统一的链路、日志、调试、部署方式

4) How：怎么落地（推荐路径）？

用 LCEL 先做“可控单链”（Prompt -> LLM -> Parser）。
加上结构化输出和错误处理（Parser + Retry）。
接入 RAG（Retriever + 引用来源）。
再上 Agent（仅在确实需要多工具决策时）。
最后做监控与部署（Callback/LangSmith + LangServe/FastAPI）。

LangChain 基础架构（最小闭环）

从一次请求到一次响应，通常经过这几层：

Input：用户问题 + 上下文（可选）
Prompt Layer：模板拼装（系统指令、历史、检索内容）
Reasoning/Tool Layer：模型推理，必要时调用工具
Knowledge Layer：Retriever 从向量库取证据
Output Layer：解析为字符串或结构化 JSON/Pydantic
Observability Layer：记录 token、耗时、错误与链路

一句话记忆：LangChain 不替代模型，而是把模型“工程化”。
竞品情况与选型建议（2026 视角）

常见替代或互补方案：

LlamaIndex：数据连接与检索抽象强，RAG 体验好
Haystack：传统检索体系成熟，企业搜索场景扎实
Semantic Kernel：偏“企业编排 + 插件”风格
AutoGen / CrewAI：多 Agent 协作范式更突出
Dify / Flowise：低代码编排快，适合业务验证
Vercel AI SDK：前端/全栈流式体验非常顺手

简化选型建议：

你要“Python 工程化 + 细粒度可控编排”：优先 LangChain。
你要“重 RAG 数据管道”：LangChain + LlamaIndex 可互补。
你要“低代码快速验证”：先 Dify/Flowise，再迁移到 LangChain。
你要“复杂多智能体协作”：LangGraph / AutoGen / CrewAI 对比评估。

Demo 01 · LCEL 语法入门 — 链式调用

本节要学什么？

LCEL（LangChain Expression Language）是 LangChain 1.0 的核心语法——它让你像拼积木一样，用 | 管道符把多个组件串联起来组成 Chain。理解 LCEL，就理解了 LangChain 1.0 的一切。
完整演示

# ========== LangChain 1.0 核心：LCEL 管道语法 ==========
# 官方文档：https://python.langchain.com/docs/expression_language/
from langchain_core.prompts import ChatPromptTemplate # 聊天提示词模板（LangChain 1.0 统一用这个）
from langchain_core.output_parsers import StrOutputParser # 把 LLM 输出转成字符串
from langchain_openai import ChatOpenAI # OpenAI 聊天模型（langchain-openai 包）
# --- 第 1 步：创建模型（temperature=0.7，控制随机性，越低越确定性）---
llm = ChatOpenAI(
model="gpt-4o-mini", # 模型名称（gpt-4o-mini 便宜速度快，效果也不错）
temperature=0.7, # 0.0=完全确定，2.0=完全随机，建议 0.0~1.0
api_key="sk-xxxxxxxx", # 你的 API Key（生产环境建议用环境变量）
base_url="https://api.openai.com/v1", # API 端点（兼容其他 OpenAI 兼容接口）
)
# --- 第 2 步：创建 Prompt 模板 ---
# PromptTemplate 的作用：把固定的结构和动态变量分开，方便复用和维护
# {topic} 是占位符，调用时会自动替换成实际传入的值
prompt = ChatPromptTemplate.from_template(
"你是一位幽默的科普作家。请用 3 句话介绍 {topic}，最后加一个彩蛋笑话。"
)
# --- 第 3 步：组装 Chain（LCEL 核心语法）---
# chain = prompt | llm | output_parser
# 管道符 | 的含义：把左边组件的输出传给右边组件的输入
# prompt.format(topic=xxx) 的结果 → llm.invoke(xxx) → output_parser.invoke(xxx)
chain = prompt | llm | StrOutputParser()
# --- 第 4 步：运行 Chain（两种方式）---
# 方式 A：直接传入字典（最常用）
result = chain.invoke({"topic": "LangChain 框架"})
print(result)
# 方式 B：先 format 再 invoke（分步执行，便于调试）
formatted_prompt = prompt.format(topic="Python 装饰器")
result2 = chain.invoke(formatted_prompt)
print(result2)
# --- 扩展：带配置的调用（stream / batch / async）---
# 批量调用（一次传入多个输入，串行执行）
multi_results = chain.batch([
{"topic": "量子计算"},
{"topic": "区块链"},
{"topic": "神经网络"}
])
for r in multi_results:
print("---", r, "---\n")

复制代码

逐行解析

行号内容逐词解释作用1# ==========#=注释分隔线（美观用）在代码中标记大段分节3from langchain_core.prompts import ChatPromptTemplatelangchain_core=LangChain核心模块, prompts=提示词子模块, ChatPromptTemplate=聊天提示模板类从 LangChain 核心包导入提示模板（1.0 版本统一用 ChatPromptTemplate）4from langchain_core.output_parsers import StrOutputParseroutput_parsers=输出解析器, StrOutputParser=字符串解析器把 LLM 返回的 AIMessage 对象转成纯字符串5from langchain_openai import ChatOpenAIlangchain_openai=OpenAI集成包, ChatOpenAI=OpenAI聊天模型类导入调用 OpenAI GPT 系列模型的客户端8llm = ChatOpenAI(...)ChatOpenAI=OpenAI聊天模型类实例化创建 LLM 实例（和 API 通信的实际对象）9model="gpt-4o-mini"model=模型名称参数, "gpt-4o-mini"=具体模型名指定用哪个模型（影响质量、速度、价格）10temperature=0.7temperature=温度参数（控制随机性）0.0=每次完全相同，2.0=极度发散，建议 0.0~1.011api_key="sk-xxxxxxxx"api_key=API密钥参数认证凭证（生产环境用环境变量，不要硬编码）12base_url="https://..."base_url=API服务器地址兼容 OpenAI API 格式的第三方接口（如硅基流动、火山引擎等）16prompt = ChatPromptTemplate.from_template(...)ChatPromptTemplate.from_template=从字符串模板创建提示对象传入模板字符串，自动解析 {topic} 等变量18{topic}{topic}=变量占位符运行时会被实际值替换掉（模板 + 变量 = 最终 Prompt）22chain = prompt | llm | StrOutputParser()| = LCEL 管道运算符（核心！把组件串联起来）把三个组件串成一条处理流水线：prompt → llm → parser25chain.invoke({"topic": "..."}).invoke()=同步调用方法, {"topic": ...}=输入字典用字典传入所有变量，触发整条 Chain 执行（同步阻塞式）29formatted_prompt = prompt.format(...).format()=只执行 Prompt 部分，不往下走把模板套上变量，但不到 LLM（用于调试看最终 Prompt 什么样）30result2 = chain.invoke(formatted_prompt)直接传入字符串（跳过 format 步骤）上一步生成了 ChatPromptValue，这里直接送入 llm35multi_results = chain.batch([...]).batch()=批量调用方法同时提交多个任务，串行执行，返回列表（适合批量处理文档等场景）常见坑

Prompt 变量名和 invoke() 传参键不一致，运行时会直接报错。
在代码里硬编码 api_key，后续很容易泄露到仓库或日志。
把 temperature 设太高，导致教程结果难复现。

生产建议

固定基础模型和温度，先保证可复现，再做 A/B 调参。
统一用环境变量注入密钥，避免在源码里出现 sk-。
所有链都先加 StrOutputParser()，减少消息对象类型差异带来的坑。

最小可运行命令

uv add langchain langchain-openai python-dotenv
uv run python demo01_lcel.py

复制代码

Demo 02 · Prompt Template 详解 — 灵活设计提示词

本节要学什么？

Prompt（提示词）是 LLM 应用的核心。LangChain 1.0 提供了多种 Prompt 模板：ChatPromptTemplate（聊天型）、PromptTemplate（纯文本型）、MessagesPlaceholder（动态消息历史）。设计好 Prompt，LangChain 才有灵魂。
完整演示

from langchain_core.prompts import (
ChatPromptTemplate, # 聊天型模板（传入消息列表）
PromptTemplate, # 纯文本模板（传入单个字符串变量）
MessagesPlaceholder, # 动态消息占位符（LangChain Memory 的关键）
HumanMessagePromptTemplate, # 用户消息模板
SystemMessagePromptTemplate, # 系统消息模板
)
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
# ========== 方式 A：纯文本 PromptTemplate（简单场景）==========
# 最基础的模板：变量替换 + 固定前缀/后缀
simple_template = PromptTemplate.from_template(
"请把以下中文翻译成英文：\n\n{text}",
# 可选：给模板加名称和描述（方便调试和日志追踪）
partial_variables={"language": "English"}, # partial_variables=预填充部分变量
)
# partial_variables：先填充 language="English"，调用时只需传 text
prompt = simple_template.partial(language="English")
result = prompt.format(text="LangChain 让 AI 开发变得简单有趣")
print(result)
# ========== 方式 B：聊天型 ChatPromptTemplate（推荐，结构更清晰）==========
# LangChain 把 LLM 对话建模为：System（系统设定）+ Human（用户）+ AIMessage（AI回复）
# 这种建模方式让模型"理解"自己在对话中的角色
chat_prompt = ChatPromptTemplate.from_messages([
# SystemMessage：给 AI 定人设（最重要，决定模型行为）
SystemMessage(
content=(
"你是一位{profession}专家。\n"
"你的工作风格是：{style}\n"
"回答问题时，总是用emoji增加趣味性。"
),
additional_kwargs={"profession": None} # additional_kwargs=附加元数据（LangChain内部用）
),
# MessagesPlaceholder：在这里动态插入【消息历史】（实现多轮对话的关键！）
# variable_name="chat_history" 必须和 invoke 时传入的键名一致
MessagesPlaceholder(variable_name="chat_history", optional=True), # optional=True=可省略
# HumanMessagePromptTemplate：用户输入（每个 invoke 都会新建一个）
HumanMessagePromptTemplate.from_template("{user_input}"),
])
# 渲染后的完整 Prompt（调试用）
print("=== 模板结构 ===")
print(chat_prompt.messages)
print("=== 变量列表 ===")
print(chat_prompt.input_variables) # 查看需要传入哪些变量
# ========== 调用方式 ==========
messages = chat_prompt.format_messages(
profession="Python",
,
user_input="什么是 LCEL？",
chat_history=[ # 传入历史消息，实现多轮对话上下文
HumanMessage(content="你好，我想学 LangChain"),
AIMessage(content="嗨！很高兴你想学 LangChain。它是一个 LLMs 应用开发框架...")
]
)
response = llm.invoke(messages)
print("AI 回复：", response.content)
# ========== 方式 C：PipelinePrompt（模板嵌套，适合复杂提示词）==========
from langchain_core.prompts import PipelinePromptTemplate
# 先定义多个子模板
introduction = PromptTemplate.from_template(
"你是 {character}，一个乐于助人的 AI 助手。"
)
main_prompt = PromptTemplate.from_template(
"{introduction}\n\n用户问：{question}\n\n你的回答："
)
# 最终模板：把子模板的结果汇入主模板
full_prompt = PipelinePromptTemplate(
pipeline_prompts=[
("introduction", introduction), # ("变量名", 模板) 元组列表
("main", main_prompt),
],
final_prompt=PromptTemplate.from_template("{main}") # 最终输出这个模板的结果
)
final = full_prompt.format(character="LangChain 助手", question="什么是 RAG？")
print(final)

复制代码

逐词解析

行号内容逐词解释作用1from langchain_core.prompts import (...)langchain_core.prompts=LangChain核心提示词模块导入所有提示词相关组件12PromptTemplate.from_template(...)from_template=从字符串创建模板的类方法最简单的模板创建方式（够用 90% 场景）14partial_variables={"language": "English"}partial_variables=预填充变量（先填一部分，剩下的调用时再填）把模板先"部分实例化"，后续只需填剩余变量15prompt = simple_template.partial(language="English").partial()=返回新模板（已预填部分变量）调用时不传 language，只传 text23ChatPromptTemplate.from_messages([...])from_messages=从消息列表创建聊天模板最灵活的模板创建方式，支持 System/Human/AI 多角色26SystemMessage(content=...)SystemMessage=系统消息类（定义AI人设）给 AI 定角色和规则（最重要的提示词组件）31MessagesPlaceholder(variable_name="chat_history", ...)MessagesPlaceholder=动态消息占位符运行时把消息历史插入到这里（多轮对话的核心）32optional=Trueoptional=占位符是否可选（不传时不报错）True=调用时不传 chat_history 也不报错38HumanMessagePromptTemplate.from_template("{user_input}")from_template=从字符串生成用户消息模板每次 invoke 新建一个用户消息（不要在模板里放死的用户消息）46chat_prompt.format_messages(...).format_messages()=格式化消息列表（返回 Message 对象列表）把模板 + 变量渲染成 LLM 可以直接吃的消息列表48HumanMessage(content="...")HumanMessage=用户消息类手动构造用户消息（Session 历史中记录用户说了什么）49AIMessage(content="...")AIMessage=AI回复消息类手动构造 AI 回复（记录 AI 说了什么，用于上下文）60PipelinePromptTemplatePipelinePrompt=流水线模板（子模板嵌套）先渲染子模板，把结果汇入主模板（适合超复杂提示词）61pipeline_prompts=[("introduction", introduction), ...]pipeline_prompts=流水线模板列表, ("变量名", 模板)=元组每个元组定义一个子模板，输出会注入到下一个模板的 {变量名} 中常见坑

MessagesPlaceholder 的变量名与输入键不一致（如 chat_history vs history）。
在 SystemMessage 里塞过多业务细节，导致 Prompt 难维护。
模板层数过深但无命名规范，调试时很难定位错误。

生产建议

用“角色/约束/输出格式”三段式组织系统提示词。
关键模板加版本号（如 prompt_v1, prompt_v2）便于回滚。
在调用前打印一次 prompt.input_variables 做自检。

最小可运行命令

uv add langchain langchain-openai
uv run python demo02_prompt_template.py

复制代码

Demo 03 · 记忆（Memory）— 让 AI 记住对话历史

本节要学什么？

没有 Memory 的 Chain 是一个"金鱼"——每次对话都不知道之前聊了什么。LangChain 1.0 提供了多种 Memory 方案：ConversationBufferMemory（最简单的历史）、ConversationSummaryMemory（AI 帮你摘要，节省 token）、VectorStoreRetrieverMemory（向量检索记忆，适用长对话）。
完整演示

from langchain_openai import ChatOpenAI
from langchain.chains import ConversationChain # 对话链（内置 Memory 集成）
from langchain.memory import (
ConversationBufferMemory, # 原始历史（最简单，Token 消耗大）
ConversationSummaryMemory, # AI 摘要历史（省 Token，推荐）
ConversationBufferWindowMemory, # 窗口记忆（只保留最近 N 条）
)
from langchain_core.messages import get_buffer_string # 把消息历史转成字符串
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
# ========== 方式 A：ConversationBufferMemory（原始历史，简单直接）==========
# ConversationBufferMemory：原封不动地存所有消息
# 优点：信息完整缺点：对话一长，Token 爆炸（GPT-4o-mini 128k 上下文但贵）
buffer_memory = ConversationBufferMemory(
return_messages=True, # return_messages=True=返回 Message 对象列表（给 LCEL 用）
# return_messages=False=返回普通字符串（给传统 Chain 用）
ai_prefix="助手", # AI 消息的前缀（影响 Prompt 里的名字）
human_prefix="用户", # 人类消息的前缀
)
# 模拟两轮对话
buffer_memory.save_context(
{"input": "我叫布鲁斯，是一名技术合伙人"}, # 存入用户说的
{"output": "你好布鲁斯！很高兴认识你，一位技术背景的合伙人。请问有什么我可以帮你的？"} # 存入AI回复
)
buffer_memory.save_context(
{"input": "我想用 LangChain 开发一个客服机器人"},
{"output": "很棒的选择！LangChain 是目前最流行的 LLM 应用框架。请问你需要什么水平的对话能力？"}
)
# 读取全部历史（返回 Message 对象列表，LCEL 的 MessagesPlaceholder 直接用）
history = buffer_memory.chat_memory.get_messages()
print("=== Buffer Memory 全部消息 ===")
for msg in history:
print(f"[{msg.type}]: {msg.content}")
# ========== 方式 B：ConversationSummaryMemory（AI 摘要，省 Token）==========
# 适用场景：对话很长（比如 20+ 轮），用原始历史太贵
# 原理：每隔几轮，AI 自动把历史"压缩"成摘要，存起来
summary_memory = ConversationSummaryMemory(
llm=llm, # 必须传入 llm（因为要调用 AI 做摘要）
return_messages=True,
buffer="", # 初始 buffer 为空（第一轮对话后自动填充摘要）
)
# 模拟长对话（10 轮）
for i in range(10):
user_input = f"这是第{i+1}轮对话，用户在说一些关于 LangChain 的事情"
ai_output = f"AI 在第{i+1}轮做出了回复"
summary_memory.save_context({"input": user_input}, {"output": ai_output})
# 查看摘要（你会发现 AI 自动把10轮对话压缩成了一段话）
print("=== Summary Memory 摘要 ===")
print(summary_memory.buffer)
# ========== 方式 C：ConversationBufferWindowMemory（窗口记忆）==========
# 适用场景：只想记住"最近 N 条"消息，不需要全部历史
window_memory = ConversationBufferWindowMemory(
k=3, # k=窗口大小，只保留最近 3 轮对话
return_messages=True,
ai_prefix="助手",
human_prefix="用户",
)
for i in range(8):
window_memory.save_context({"input": f"用户第{i}轮"}, {"output": f"AI第{i}轮回复"})
print("=== 窗口 Memory（只看最后3条）===")
print(window_memory.buffer) # 只有第5、6、7轮的内容了
# ========== 把 Memory 接入 LangChain 1.0 LCEL Chain ==========
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# LangChain 1.0 推荐做法：在 Prompt 的 MessagesPlaceholder 插入 Memory
prompt_with_memory = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个友好的 AI 助手，名字叫小 L。请基于对话历史回答问题。"),
MessagesPlaceholder(variable_name="history", optional=True),
HumanMessagePromptTemplate.from_template("{question}"),
])
# 用 LCEL 手动组装带 Memory 的 Chain
def get_history_str(memory):
"""把 Memory 对象中的历史转成字符串，注入到 prompt"""
messages = memory.chat_memory.get_messages()
return get_buffer_string(messages) # 把 Message 对象列表转成可读字符串
# 注意：LangChain 1.0 的 LCEL 方式需要手动把 history 传入 prompt
chain = prompt_with_memory | llm | StrOutputParser()
# 模拟多轮对话（每次都把历史传进去）
memory = ConversationBufferMemory(
return_messages=False, # return_messages=False → get_buffer_string() 返回字符串
ai_prefix="助手",
human_prefix="用户"
)
memory.save_context({"input": "我叫布鲁斯"}, {"output": "你好布鲁斯！"})
memory.save_context({"input": "我喜欢 Python 编程"}, {"output": "Python 是一门很棒的语言！"})
# 调用时把 history 传入字典
result = chain.invoke({
"question": "你还记得我叫什么名字吗？",
"history": get_buffer_string(memory.chat_memory.get_messages()) # 把历史作为字符串传入
})
print("AI 回复：", result)

复制代码

逐行解析

行号内容逐词解释作用1from langchain.chains import ConversationChainConversationChain=内置对话链（已集成 Memory）LangChain 1.0 前的传统方式（现在更推荐用 LCEL 手动组装）3from langchain.memory import ConversationBufferMemorymemory=记忆模块, ConversationBufferMemory=原始消息历史记忆把所有对话原封不动存起来6return_messages=Truereturn_messages=返回格式控制参数, True=返回Message对象LCEL 的 MessagesPlaceholder 需要 Message 对象列表7ai_prefix="助手"ai_prefix=AI消息前缀（显示名）影响 Prompt 里 AI 消息的标记方式（中文场景友好）17buffer_memory.save_context(...).save_context()=保存对话上下文方法, {"input": ..., "output": ...}=用户+AI对把一对问答存进 Memory（相当于 append 到历史列表）23history = buffer_memory.chat_memory.get_messages().chat_memory=聊天记录管理器, .get_messages()=获取所有消息列表从 Memory 中取出完整的消息对象列表35ConversationSummaryMemory(llm=llm, ...)llm=必须传入一个 LLM 实例（用于生成摘要）AI 摘要记忆：每隔几轮调用 LLM 自动压缩历史37buffer=""buffer=初始历史（空字符串）第一轮对话后，AI 会自动生成第一段摘要填入这里47ConversationBufferWindowMemory(k=3, ...)k=窗口大小参数只保留最近 k 条对话（更省 Token）64MessagesPlaceholder(variable_name="history", optional=True)variable_name=变量名（必须和 invoke 字典的 key 一致）动态插入对话历史的地方（LCEL Memory 集成的关键）69get_buffer_string(messages)get_buffer_string=把消息列表转成可读字符串的工具函数LangChain 提供的格式化工具，把 Message 对象列表转成 "\n\nHuman: ...\nAI: ..." 格式78memory.chat_memory.get_messages().chat_memory=底层 BaseChatMessageHistory 对象获取 Memory 里存的所有消息（用于传给 get_buffer_string）常见坑

不限制历史长度，Token 成本会快速失控。
Memory 存了历史，但调用链没把历史传入 Prompt。
把“长期偏好”和“短期会话”混在同一个 Memory 中，语义污染。

生产建议

默认从窗口记忆开始（如近 6~10 轮），再评估是否要摘要记忆。
长会话建议“窗口 + 摘要”组合，而不是只存原文。
历史写入要有脱敏策略，避免日志里留个人隐私或密钥。

最小可运行命令

uv add langchain langchain-openai
uv run python demo03_memory.py

复制代码

Demo 04 · Output Parser — 让 LLM 返回结构化数据

本节要学什么？

LLM 默认只返回纯文本。但在生产环境，你需要 JSON、CSV、Python 对象。Output Parser 就是把 LLM 的"自由文本"变成"结构化数据"的转换器。LangChain 1.0 内置了 JsonOutputParser、PydanticOutputParser、CommaSeparatedListOutputParser 等。
完整演示

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import (
JsonOutputParser, # 把 LLM 输出解析成 JSON 字典（最常用）
PydanticOutputParser, # 把 LLM 输出解析成 Pydantic 模型（最严格，推荐生产用）
CommaSeparatedListOutputParser, # 把 LLM 输出解析成 Python 列表
)
from langchain.output_parsers import RetryOutputParser, OutputFixingParser # 纠错解析器
from langchain_core.runnables import RunnablePassthrough
from pydantic import BaseModel, Field, field_validator # Pydantic：Python 数据验证库
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
# ========== 方式 A：JsonOutputParser（简单 JSON 解析）==========
# 定义想要的 JSON 结构（用自然语言描述给 LLM）
json_parser = JsonOutputParser()
print("=== LLM 需要生成的格式 ===")
print(json_parser.get_format_instructions()) # 打印 LLM 看见的格式说明
# 创建模板（把格式说明自动注入到 Prompt 里）
json_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个数据提取助手。"),
HumanMessagePromptTemplate.from_template(
"从以下文本中提取信息，以 JSON 格式返回：\n\n{text}\n\n"
"请严格按照以下格式返回：\n{format}"
),
])
# 自动把 {format} 替换成 JsonOutputParser 的格式说明
chain_json = json_prompt | llm | json_parser
result = chain_json.invoke({
"text": "布鲁斯，37岁，来自深圳，是一名技术合伙人，擅长 Python 和 Java。"
})
print("=== JSON 解析结果 ===")
print(result) # {'name': '布鲁斯', 'age': '37', 'city': '深圳', ...}
print(result["name"]) # 布鲁斯（字典键访问）
# ========== 方式 B：PydanticOutputParser（最推荐，数据验证）==========
# Pydantic 模型：比 JSON Schema 更强大的数据验证（Python 版 TypeScript）
# 优势：类型自动转换、默认值、字段验证、超长报错信息
class PersonInfo(BaseModel):
"""人物信息模型"""
name: str = Field(description="人物姓名") # description=L给LM看的字段说明
age: int = Field(description="人物年龄（必须是整数）")
city: str = Field(description="所在城市")
skills: list[str] = Field(default_factory=list, description="掌握的技能列表")
@field_validator("age")
@classmethod
def age_must_be_positive(cls, v: int) -> int:
"""自定义验证器：年龄必须是正数"""
if v <= 0 or v > 150:
raise ValueError(f"年龄 {v} 不合理！")
return v
# 创建 Pydantic 解析器
pydantic_parser = PydanticOutputParser(pydantic_object=PersonInfo)
# 方式 1：把格式说明注入 Prompt（标准做法）
pydantic_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个数据提取助手。"),
HumanMessagePromptTemplate.from_template(
"从以下文本中提取信息：\n\n{text}\n\n{format}"
),
])
pydantic_prompt = pydantic_prompt.partial(format=pydantic_parser.get_format_instructions())
# 方式 2：用 .from_langchain() 快速创建（LangChain 1.0 新增！）
# prompt = pydantic_parser.get_default_prompt() # 自动包含格式说明
chain_pydantic = pydantic_prompt | llm | pydantic_parser
person: PersonInfo = chain_pydantic.invoke({
"text": "布鲁斯，37岁，深圳技术合伙人，擅长 Python、Java、Go。"
})
print("=== Pydantic 解析结果 ===")
print(f"姓名：{person.name}") # 布鲁斯（直接对象属性访问，比字典更优雅）
print(f"年龄：{person.age}") # 37（字符串 "37" 自动转成 int）
print(f"城市：{person.city}") # 深圳
print(f"技能：{person.skills}") # ['Python', 'Java', 'Go']
# ========== 方式 C：RetryOutputParser（自动纠错）==========
# 当 LLM 返回的格式有瑕疵时（缺字段、多字段），自动让 LLM 重新生成
# 原理：解析失败 → 调用 OutputFixingParser 让另一个 LLM 修复
base_parser = JsonOutputParser()
retry_parser = RetryOutputParser.from_llm(
parser=base_parser, # 基础解析器
llm=llm, # 用来修复错误的 LLM（可以用更便宜的模型）
max_retries=3, # 最多重试 3 次
)
prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="提取以下文本中的信息，以 JSON 返回。只返回 JSON，不要其他内容。"),
HumanMessagePromptTemplate.from_template("{text}"),
])
chain_with_retry = prompt | llm | retry_parser
# 即使 LLM 返回了不完美的 JSON，RetryOutputParser 也会尝试修复
# ========== 方式 D：CommaSeparatedListOutputParser（列表解析）==========
list_parser = CommaSeparatedListOutputParser()
list_prompt = ChatPromptTemplate.from_messages([
HumanMessagePromptTemplate.from_template("列出 {subject} 的 {n} 个优点，用逗号分隔。"),
])
list_chain = list_prompt | llm | list_parser
result = list_chain.invoke({"subject": "Python", "n": 5})
print("=== 列表解析结果 ===")
print(result) # ['优点1', '优点2', '优点3', '优点4', '优点5']
print(result[0]) # 第一个元素

复制代码

逐行解析

行号内容逐词解释作用1from pydantic import BaseModel, Field, field_validatorpydantic=数据验证库（FastAPI 的底层依赖）, BaseModel=Pydantic模型基类, Field=字段定义器, field_validator=字段验证器导入 Pydantic 三大核心工具（比 JSON Schema 更好用的数据验证）11JsonOutputParser()JsonOutputParser=JSON输出解析器把 LLM 返回的文本解析成 Python 字典12json_parser.get_format_instructions().get_format_instructions()=获取格式说明字符串打印 LLM 需要的输出格式描述（一般是一段自然语言指令）16HumanMessagePromptTemplate.from_template("{text}\n{format}"){text}=用户文本占位符, {format}=解析器格式说明占位符把用户输入和格式说明一起拼进 Prompt22chain_json = json_prompt | llm | json_parserLCEL 三件套：模板 → 模型 → 解析器完整的处理流水线（Prompt 格式化 → LLM → 解析成字典）25result["name"]result=Python 字典, ["name"]=键访问JsonOutputParser 返回的是 dict，直接用键访问30class PersonInfo(BaseModel)class=定义类, PersonInfo=类名（自定义）, BaseModel=Pydantic模型基类定义一个 Pydantic 模型（声明字段、类型、验证规则）31name: str = Field(description="...")name=字段名（蛇命名）, str=字段类型, Field(...)=字段元数据定义字符串字段，description 是给 LLM 看的说明（决定 LLM 填什么）32skills: list[str] = Field(default_factory=list, ...)default_factory=list=默认值工厂（每次创建新实例时调用 list()）如果 LLM 没返回 skills，就默认返回空列表（避免 None）34@field_validator("age")@field_validator=字段验证器装饰器, "age"=作用于 age 字段声明一个验证方法，检查 age 是否合法35def age_must_be_positive(cls, v: int) -> int:cls=类本身（类方法默认参数）, v=字段实际值, -> int=返回类型自定义验证逻辑：年龄必须在 1~150 之间36raise ValueError(...)raise=抛出异常, ValueError=值错误类型不满足条件时报错，Pydantic 会自动把这个字段标记为错误45pydantic_parser = PydanticOutputParser(pydantic_object=PersonInfo)pydantic_object=传入 Pydantic 模型类（不是实例！）创建 Pydantic 解析器（会自动调用 PersonInfo 模型验证数据）51pydantic_prompt.partial(format=pydantic_parser.get_format_instructions()).partial()=预填充变量, format=已填充变量名自动把格式说明注入 Prompt（LangChain 推荐做法）58person: PersonInfo = chain_pydantic.invoke(...)person=PersonInfo 类型标注, PersonInfo=自定义 Pydantic 类解析结果直接是 PersonInfo 对象，有类型提示（IDE 自动补全友好）60person.nameperson=PersonInfo 对象, .name=属性访问直接用属性访问（比 dict 更优雅，错误时 IDE 会有提示）65RetryOutputParser.from_llm(...)RetryOutputParser=重试解析器, from_llm=类方法（从 LLM 创建）当基础解析器失败时，自动用 LLM 修复输出格式66max_retries=3max_retries=最大重试次数连续失败 3 次后放弃（防止无限循环）76CommaSeparatedListOutputParser()CommaSeparatedListOutputParser=逗号分隔列表解析器把 "a, b, c" 这样的字符串解析成 ["a", "b", "c"]常见坑

只让模型“返回 JSON”，但没有把格式约束注入 Prompt。
解析失败后直接报错退出，没有重试或修复策略。
Pydantic 模型字段定义不清晰，导致模型输出漂移。

生产建议

优先 PydanticOutputParser，让 schema 成为契约。
对关键链路增加 RetryOutputParser 或失败兜底分支。
给解析失败做可观测记录（原始输出、异常、重试次数）。

最小可运行命令

uv add langchain langchain-openai pydantic
uv run python demo04_output_parser.py

复制代码

Demo 05 · RAG 核心 — Document Loader + Text Splitter + Vector Store

本节要学什么？

RAG（Retrieval-Augmented Generation）= 检索 + 生成。LangChain 是 RAG 的最佳载体。本 Demo 讲清楚 RAG 的第一步：如何把文档（PDF、TXT、网页）加载进来，切块（Chunk），然后存入向量数据库。这是所有 RAG 应用的地基。
完整演示

# ========== RAG 第一步：文档加载（Document Loader）==========
from langchain_community.document_loaders import (
TextLoader, # 加载纯文本文件（.txt, .md）
PDFLoader, # 加载 PDF 文件（需要 pip install pypdf）
WebBaseLoader, # 从网页 URL 加载内容
UnstructuredMarkdownLoader, # 更智能的 Markdown 加载器
)
from langchain_text_splitters import (
RecursiveCharacterTextSplitter, # 按字符递归分割（最常用，效果好）
CharacterTextSplitter, # 按单个字符分割（简单场景）
MarkdownTextSplitter, # 按 Markdown 语法分割（保留标题结构）
LanguageTextSplitter, # 按编程语言语法分割（代码文档专用）
)
from langchain_openai import OpenAIEmbeddings # 文本向量化模型
from langchain_community.vectorstores import Chroma # 向量数据库（轻量，支持本地）
# --- 加载文档 ---
# 1. TextLoader：加载本地文本文件
# 假设同目录下有个 essay.txt
loader = TextLoader("essay.txt", encoding="utf-8")
documents = loader.load()
print(f"加载文档数: {len(documents)}")
print(f"第一段内容（前100字）: {documents[0].page_content[:100]}...")
print(f"元数据: {documents[0].metadata}") # metadata=文档来源、时间等附加信息
# 2. PDFLoader：加载 PDF（每页单独作为一个 Document）
pdf_loader = PDFLoader("paper.pdf", encoding="utf-8")
pdf_docs = pdf_loader.load()
print(f"PDF 页数: {len(pdf_docs)}")
# 3. WebBaseLoader：从网页加载（适合让 AI 总结文章）
web_loader = WebBaseLoader("https://python.langchain.com/docs/introduction")
web_docs = web_loader.load()
print(f"网页内容（前200字）: {web_docs[0].page_content[:200]}")
# --- 文档切块（Text Splitter）---
# 切块的原因：LLM 有上下文窗口限制（不能把整本书塞进去），
# 而且小块检索更精准（相关段落 vs 整本书）
# RecursiveCharacterTextSplitter：按优先级依次尝试分割（["\n\n", "\n", " ", ""]）
# 优点：尽量保持语义完整（段落 > 句子 > 词），不会把一句话切成两半
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # chunk_size=每块最大字符数（建议 300~1000）
chunk_overlap=50, # chunk_overlap=块与块之间的重叠字符数（ chunk_overlap=50, # chunk_overlap=块与块之间的重叠字符数（建议 chunk_size 的 10%~20%）
length_function=len, # length_function=计算块大小的函数（默认 len=字符数）
add_start_index=True, # add_start_index=True=在元数据里记录每块在原文中的起始位置
)
# 执行切块（返回 Document 列表，每个 Document = 一块文本 + 元数据）
chunks = text_splitter.split_documents(documents)
print(f"切块数量: {len(chunks)}")
print(f"第一块（前100字）: {chunks[0].page_content[:100]}")
print(f"元数据（含原文位置）: {chunks[0].metadata}")
# 其他切块器（按场景选用）
# MarkdownTextSplitter：按 Markdown 标题/段落切块（适合 md 文件）
md_splitter = MarkdownTextSplitter(chunk_size=300, chunk_overlap=30)
md_chunks = md_splitter.split_text(open("readme.md", encoding="utf-8").read())
# LanguageTextSplitter：按编程语言语法切块（适合代码文档，保留函数/类边界）
from langchain_text_splitters import Language
python_splitter = LanguageTextSplitter(language=Language.PYTHON, chunk_size=500, chunk_overlap=50)
# --- 向量数据库：存储 + 检索 ---
# OpenAI Embeddings：把文本转成 1536 维向量（GPT 系列的Embedding模型）
embeddings = OpenAIEmbeddings(
model="text-embedding-3-small", # text-embedding-3-small=新版更快更便宜，1536维
api_key="sk-xxxxxxxx"
)
# Chroma：轻量级向量数据库（纯 Python，支持本地文件存储，开发调试首选）
# 生产环境可用 FAISS（Facebook 出品，亿级向量）、Pinecone（云服务）、Milvus（国产开源）
vectorstore = Chroma.from_documents(
documents=chunks, # 要存储的文档块列表
embedding=embeddings, # 用什么向量化模型
persist_directory="./chroma_db", # 持久化到本地目录（重启后数据不丢失）
)
# 保存到磁盘（Chroma 需要手动调用 persist）
vectorstore.persist()
# --- 向量检索：查询最相关的 K 个文档块 ---
# similarity_search：基于语义相似度查找最相关的文档
# query=搜索的文本（不需要完全匹配，Embedding 会做语义理解）
query = "LangChain 的核心概念是什么？"
relevant_docs = vectorstore.similarity_search(query=query, k=3)
# k=返回最相似的 3 个文档块
print(f"检索到 {len(relevant_docs)} 个相关文档块：")
for i, doc in enumerate(relevant_docs, 1):
print(f"\n--- 相关文档 {i}（相似度相关）---")
print(doc.page_content[:200])
# similarity_search_with_score：同时返回相似度分数（0=完全相同，越小越相似）
relevant_docs_with_scores = vectorstore.similarity_search_with_score(query=query, k=3)
for doc, score in relevant_docs_with_scores:
print(f"\n[分数: {score:.4f}] {doc.page_content[:150]}")
# mmr（最大边际相关性）：检索时兼顾多样性，避免返回重复内容
mmr_docs = vectorstore.max_marginal_relevance_search(query=query, k=3, fetch_k=10)
# fetch_k=先从向量库取出 10 个候选，再从中选 3 个多样化的
# --- 创建 Retriever（LangChain 1.0 推荐用法）---
# VectorStoreRetriever：把向量库包装成 Retriever 接口
retriever = vectorstore.as_retriever(
search_type="similarity", # similarity=语义相似度（最常用）
search_kwargs={"k": 3} # k=每次返回 3 个最相关块
)
# 使用 Retriever（比直接调 VectorStore 更规范，可被 Chain 直接用）
docs = retriever.invoke("LangChain 是什么？")
print(f"Retriever 返回 {len(docs)} 个文档块")

复制代码

逐行解析

行号内容逐词解释作用1chunk_overlap=50chunk_overlap=块间重叠字符数（保持上下文连续性）50字符的重叠让相邻块有上下文衔接（不会在切分处丢失信息）3length_function=lenlength_function=计算文本长度的函数len=按字符数计（也可以自定义，比如按 token 数计）4add_start_index=Trueadd_start_index=记录原文起始位置方便后续追踪这段文本来自原文哪个位置6chunks = text_splitter.split_documents(documents).split_documents()=切分文档列表把 Document 对象列表切成更小的 Document 块列表13MarkdownTextSplitterMarkdownTextSplitter=按 Markdown 语法切块保留 # 标题、## 副标题等结构，适合 md 文档17LanguageTextSplitter(language=Language.PYTHON, ...)Language=编程语言枚举, PYTHON=Python语言按函数/类/import 等语法结构切块（适合代码文档）23embeddings = OpenAIEmbeddings(...)OpenAIEmbeddings=OpenAI向量化模型客户端把任意文本转成 1536 维浮点数向量24model="text-embedding-3-small"model=Embedding模型名新版 Embedding 模型（比 ada-002 更小更快，价格降 5 倍）29Chroma.from_documents(...)Chroma=向量数据库, .from_documents=从文档列表创建把文档块向量化后存入 Chroma，返回 VectorStore 对象31persist_directory="./chroma_db"persist_directory=持久化存储路径把向量库存到本地磁盘（否则内存存储，重启后消失）34vectorstore.persist().persist()=手动持久化（Chroma 默认自动，但显式调用更安全）把内存中的向量数据写入磁盘40vectorstore.similarity_search(query=query, k=3).similarity_search=语义相似度检索, query=查询文本, k=返回数量用 Embedding 把 query 转成向量，在向量库中找最相似的 k 个文档44similarity_search_with_score.similarity_search_with_score=带分数的检索返回 (Document, score) 元组，score 表示相似度（0=完全相同）48max_marginal_relevance_search(query, k, fetch_k)mmr=最大边际相关性（一种兼顾相关性和多样性的检索策略）先粗筛 fetch_k 个候选，再用 MMR 算法选 k 个多样化的结果54retriever = vectorstore.as_retriever(...).as_retriever()=把 VectorStore 包装成 Retriever 接口Retriever 是 LangChain 的标准检索接口（Chain 的通用输入）57retriever.invoke("...").invoke()=LCEL 统一调用方法用字符串直接检索（Retriever.invoke 等价于 similarity_search）常见坑

切块过大或过小，分别会导致召回噪声高或上下文丢失。
检索时 k 固定不调，问答质量很不稳定。
向量库更新后没有重建索引或版本管理，结果不可追溯。

生产建议

先做离线评估，找到适合你语料的 chunk_size/chunk_overlap。
检索默认返回 source 元数据，方便答案溯源。
把“数据版本 + embedding 模型版本”写入元数据。

最小可运行命令

uv add langchain langchain-openai langchain-community chromadb pypdf beautifulsoup4
uv run python demo05_rag_ingest.py

复制代码

Demo 06 · RAG Chain — 完整检索增强生成

本节要学什么？

本 Demo 是 RAG 的核心——把 Demo 05 的检索结果接进 LCEL Chain，让 LLM 基于检索到的真实文档回答问题，而不是凭空编造（Hallucination）。这是 LangChain 最多人用的场景。
完整演示

# ========== 完整 RAG Chain ==========
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7, api_key="sk-xxx")
embeddings = OpenAIEmbeddings(model="text-embedding-3-small", api_key="sk-xxx")
# --- 加载已有的向量库（不用重新建，用 Demo 05 的）---
vectorstore = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings,
)
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 3}
)
# ========== 方式 A：stuff 模式（把所有文档塞进一次 Prompt，最简单）==========
# stuff=把检索到的所有文档"塞进"一个 Prompt
# 优点：简单，Token 消耗中等缺点：文档太多时超过上下文窗口
# 适合：检索结果 < 10 个文档的场景
stuff_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content=(
"你是一个知识渊博的助手，基于提供的文档片段回答问题。\n"
"如果文档中没有答案，请直接说「没有找到相关信息」，不要编造。\n"
"每次回答都要注明信息来源。"
)),
MessagesPlaceholder(variable_name="context"), # 检索到的文档塞在这里
HumanMessagePromptTemplate.from_template("问题：{question}"),
])
def format_docs(docs: list) -> str:
"""
把 Document 对象列表格式化成字符串
LCEL 里用这个函数把 retriever 的输出格式化成字符串
"""
return "\n\n".join(
f"[来源 {i+1}] {doc.page_content}" # 加编号方便 LLM 引用
for i, doc in enumerate(docs)
)
# LCEL：检索 → format → prompt → llm → parser
rag_chain = (
{
# RunnablePassthrough：透传参数（把 question 原样传下去）
# retriever.invoke(question)：根据 question 检索相关文档
"question": RunnablePassthrough(),
"context": retriever | format_docs, # 先检索，再格式化（管道组合）
}
| stuff_prompt # 把 question + context 注入 prompt
| llm # 调用 LLM
| StrOutputParser() # 解析成字符串
)
# 运行 RAG Chain
result = rag_chain.invoke("LangChain 是什么？有哪些核心概念？")
print("=== RAG 结果 ===")
print(result)
# ========== 方式 B：refine 模式（逐文档处理，迭代优化）==========
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain
from langchain_core.runnables import RunnablePassthrough
# create_retrieval_chain：LangChain 1.0 推荐的官方 RAG Chain 构造方式
# create_stuff_documents_chain：把文档塞进 prompt（stuff 模式）
retrieval_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个问答助手。根据以下上下文回答用户问题。"),
MessagesPlaceholder(variable_name="context"),
HumanMessagePromptTemplate.from_template("问题：{input}"),
])
document_chain = create_stuff_documents_chain(
llm=llm,
prompt=retrieval_prompt,
)
retrieval_chain = create_retrieval_chain(
retriever, # 检索器
document_chain, # 处理检索结果文档的 Chain
)
response = retrieval_chain.invoke({"input": "LangChain 1.0 有哪些新特性？"})
print("=== Retrieval Chain 结果 ===")
print(f"答案：{response['answer']}")
print(f"涉及的源文档数：{len(response['context'])}")
# ========== 方式 C：map-reduce 模式（大文档集合适用）==========
from langchain.chains.combine_documents.base import create_stuff_documents_chain
from langchain.chains import create_retrieval_chain
# map_reduce：把每个文档单独喂给 LLM 生成答案，最后再汇总
# 适合：文档数量 > 10 个，单个文档很长，用 stuff 会超上下文
# 缺点：Token 消耗大（每个文档都要过一次 LLM）
# langchain 的 map_reduceDocumentsChain
from langchain.chains.combine_documents import MapReduceDocumentsChain, ReduceDocumentsChain, StuffDocumentsChain
# Map 阶段：对每个文档单独处理
map_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是文档分析助手。请根据以下文档片段回答问题。"),
HumanMessagePromptTemplate.from_template("问题：{question}\n\n文档：{context}"),
])
# Reduce 阶段：把所有 Map 的结果汇总
reduce_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是一个汇总助手。请把多个答案合并成一个简洁的回复。"),
HumanMessagePromptTemplate.from_template("多个答案：\n{context}"),
])
# ReduceDocumentsChain：把多个文档（结果）合并
reduce_doc_chain = create_stuff_documents_chain(llm=llm, prompt=reduce_prompt)
reduce_chain = ReduceDocumentsChain(
combine_docs_chain=reduce_doc_chain, # 合并策略（stuff）
collapse_doc_chain=reduce_doc_chain, # collapse=当文档太多时先合并再合并
)
# MapReduceDocumentsChain：Map → Reduce 两阶段链
map_reduce_chain = MapReduceDocumentsChain(
llm_chain=create_stuff_documents_chain(llm=llm, prompt=map_prompt), # 每个文档单独调用
combine_documents_chain=reduce_chain, # 最后合并所有答案
)
# 用 map_reduce_chain 处理超过 context 窗口的大量文档
# result = map_reduce_chain.invoke({"input": "...", "question": "..."})
# ========== RAG + Memory（带上下文的对话 RAG）==========
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# ConversationalRetrievalChain：专门做"带历史的 RAG 对话"
from langchain.chains import ConversationalRetrievalChain
# 每次回答都要参考历史对话内容（比如用户问"它是怎么工作的？"——"它"指代上一轮的"LangChain"）
condense_prompt = ChatPromptTemplate.from_messages([
SystemMessage(content="你是对话助手，负责把对话历史和最新问题合并成一个独立问题。"),
MessagesPlaceholder(variable_name="chat_history"),
HumanMessagePromptTemplate.from_template("{question}"),
])
# 合并历史 + 新问题 → 检索 → 回答
qa_chain = ConversationalRetrievalChain.from_llm(
llm=llm,
retriever=retriever,
condense_prompt=condense_prompt, # 把对话历史 + 新问题合并成检索词
memory=None, # 可传入 ConversationMemory（用 Demo 03 的 BufferMemory）
)
chat_history = [
HumanMessage(content="LangChain 支持哪些模型？"),
AIMessage(content="LangChain 支持 OpenAI、Anthropic、Google Gemini、HuggingFace 等主流 LLM 模型。"),
]
result = qa_chain.invoke({
"question": "那 embedding 模型呢？",
"chat_history": chat_history, # 传入历史，自动理解"那...呢"指代什么
})
print("AI 回复：", result["answer"])

复制代码

逐行解析

行号内容逐词解释作用1from langchain.chains.combine_documents import create_stuff_documents_chaincombine_documents=文档组合链模块导入 stuff 模式文档处理链（把多个文档塞进一个 prompt）3from langchain.chains import create_retrieval_chaincreate_retrieval_chain=检索增强生成链工厂函数LangChain 1.0 推荐的 RAG Chain 构造方式8vectorstore = Chroma(persist_directory=..., embedding_function=...)embedding_function=embedding 函数参数（Chroma 1.0 新命名）从已有目录加载向量库（而不是重新 from_documents）13stuff_prompt = ChatPromptTemplate.from_messages([...])stuff=填充模式（把所有文档塞进一次调用）把检索到的所有文档一次性塞进一个 Prompt17MessagesPlaceholder(variable_name="context")variable_name="context"=占位符变量名（必须和 format_docs 返回的键名一致）在 Prompt 里给检索文档留一个插入位置23def format_docs(docs: list) -> str:format_docs=自定义格式化函数把 Document 对象列表格式化成带编号的字符串，方便 LLM 引用来源24return "\n\n".join([...])"\n\n".join=双换行拼接, f"[来源 {i+1}]"=来源编号把多个文档用双换行拼接，每段前加 [来源 X] 编号30RunnablePassthrough()RunnablePassthrough=透传组件（把输入原样传递下去）在 LCEL 里透传 question 参数（让 question 能同时流向下游）31"context": retriever | format_docsretriever | format_docs=先检索再格式化LCEL 管道：先让 retriever 处理 question，再把结果传给 format_docs33| stuff_prompt | llm | StrOutputParser()LCEL 三件套完整 RAG Chain：拼好的字典 → 渲染 prompt → LLM → 字符串输出39rag_chain.invoke("LangChain 是什么？").invoke()=同步调用（传入字符串，RunnablePassthrough 自动处理）一句话触发完整 RAG 流程43create_stuff_documents_chain(llm=llm, prompt=retrieval_prompt)create_stuff_documents_chain=stuff模式文档链工厂函数创建处理文档列表的 Chain（把 docs 塞进 prompt）45create_retrieval_chain(retriever, document_chain)create_retrieval_chain=检索+问答链工厂创建完整 RAG Chain（检索 → 文档处理 → 回答）48response["answer"]response=返回字典, ["answer"]=答案键create_retrieval_chain 返回 dict，含 answer + context49response["context"]["context"]=检索到的源文档列表查看 RAG 用到了哪些文档（方便调试和溯源）58MapReduceDocumentsChainMapReduce=先分后合模式（Map=每个文档单独处理，Reduce=合并结果）适合文档量大的场景（避免超过上下文限制）67ReduceDocumentsChain(combine_docs_chain=..., collapse_doc_chain=...)combine_docs_chain=合并策略, collapse_doc_chain=压缩合并策略先 reduce 再 collapse（文档太多时先压缩数量再合并）72ConversationalRetrievalChain.from_llm(...)ConversationalRetrievalChain=对话检索链（专门做多轮RAG）支持"它"、"那"、"这些"等指代消解（理解上下文）74condense_prompt=condense_promptcondense_prompt=历史压缩提示词把 chat_history + 新问题合并成一个独立检索词（解决指代问题）75memory=Nonememory=可传入 ConversationMemory如果不传 memory，需在每次 invoke 时手动传 chat_history常见坑

回答阶段没附来源文档，用户无法判断答案可信度。
只测“能答对”，不测“答错时是否胡编”。
把检索上下文无脑塞满 Prompt，导致成本和延迟飙升。

生产建议

输出里始终附 sources 或 context 摘要，提升可解释性。
对高风险问题加“无答案就拒答”策略。
使用 create_retrieval_chain 保持链路标准化，便于维护。

最小可运行命令

uv add langchain langchain-openai langchain-community chromadb
uv run python demo06_rag_chain.py

复制代码

Demo 07 · Tool + Agent — 让 AI 自主调用工具

本节要学什么？

Agent（智能体）是 LangChain 的最强大特性——AI 不是一条道走到黑，而是有"思考能力"：看问题 → 决定是否调用工具 → 调用哪个工具 → 看结果 → 继续推理或给出最终答案。LangChain 1.0 用 @tool 装饰器定义工具，用 create_react_agent 构建 ReAct 推理 Agent。
完整演示

[code]# ========== 定义 Tool（LangChain 1.0 新语法）==========from langchain_core.tools import tool # @tool 装饰器（LangChain 1.0 新增）from langchain_openai import ChatOpenAIfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholderfrom langchain_core.utils.openai_functions import convert_to_openai_function_schemafrom langchain_core.output_parsers import StrOutputParserllm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)# --- 用 @tool 装饰器定义工具（LangChain 1.0 最简洁的方式）---@tooldef multiply(a: int, b: int) -> int: """ 计算两个整数的乘积。当你需要做乘法运算时使用这个工具。 Args: a: 第一个整数 b: 第二个整数 Returns: 两个整数的乘积（a * b） """ return a * b@tooldef get_weather(city: str) -> str: """ 查询指定城市的当前天气。当用户问某个城市的天气时使用这个工具。 Args: city: 城市名称（中文或英文都可以） Returns: 城市的天气情况描述字符串 """ # 模拟天气 API（实际项目替换成真实 API，如心知天气、OpenWeatherMap） weather_db = { "深圳": "☀️ 晴天，28°C，湿度 75%", "北京": "
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

账号		自动登录	找回密码
密码			立即注册

LangChain教程-2、Langchian基础

相关帖子

浏览过的版块

签约作者

LangChain教程-2、Langchian基础

相关帖子

相关推荐

浏览过的版块

签约作者