🦜🔗 像搭积木一样玩转 AI —— LangChain 核心组件入门笔记

📖 什么是 LangChain?
想象一下,你要造一个智能机器人:
- 大语言模型(比如 ChatGPT)是机器人的大脑 —— 它很聪明,但只会"说话"
- LangChain 就是机器人的身体和手 —— 它让大脑能"看"到数据、"拿"到工具、"记住"过去的事
一句话总结:LangChain 是一个"积木盒",帮你把大语言模型和各种能力拼在一起,变成一个真正能干活的 AI 应用。
┌─────────────────────────────────────────────┐
│ 你的 AI 应用 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌────────┐ │
│ │ 模型 🧠 │──▶│ 工具 🔧 │──▶│ 记忆 💾│ │
│ └──────────┘ └──────────┘ └────────┘ │
│ ▲ │ │
│ └─────── 中间件 ⚙️ ──────────────┘ │
│ │
│ 输入 ──▶ [ 模型 ──▶ 工具 ──▶ 记忆 ] ──▶ 输出 │
│ ▲ │
│ 系统提示 📋 │
│ │
└─────────────────────────────────────────────┘1️⃣ 模型(Model)—— AI 的大脑
模型就是 AI 的"大脑",负责理解和生成文字。LangChain 把模型分成两种用法:静态模型 和 动态模型。
🏠 静态模型 —— 固定一个大脑
静态模型就像你买了一台手机,型号固定了,用的时候直接拿出来用。
from langchain.chat_models import init_chat_model
# 方式一:直接指定提供商和模型名
model = init_chat_model("openai", model="gpt-4o")
# 方式二:简写(自动识别提供商)
model = init_chat_model("openai:gpt-4o")
# 直接用!就像打电话一样简单
response = model.invoke("你好,介绍一下自己")
print(response.content)🔀 动态模型 —— 运行时换大脑
动态模型就像手机可以插不同的 SIM 卡,今天用中国联通,明天换成中国移动,随时切换。
from langchain.chat_models import init_chat_model
# 创建一个"可配置"的模型 —— 先不确定用哪个
model = init_chat_model("openai:gpt-4o", configurable_fields=["model"])
# 调用时再决定用哪个模型!
response = model.invoke(
"写一首诗",
config={"configurable": {"model": "anthropic:claude-3-sonnet"}} # 动态切换!
)
print(response.content)
# 还可以加备用模型,主模型挂了自动切换(就像备胎)
robust_model = (
init_chat_model("anthropic:claude-3-5-haiku-latest")
.with_fallbacks([
init_chat_model("openai:gpt-4o-mini") # 备用模型
])
)📊 两种模型对比
| 特点 | 静态模型 | 动态模型 |
|---|---|---|
| 确定时间 | 写代码时就确定了 | 运行时才决定 |
| 灵活性 | ❌ 固定 | ✅ 随时切换 |
| 适用场景 | 简单应用、单一模型 | 多模型切换、用户自选、企业级应用 |
| 代码复杂度 | 简单 | 稍复杂 |
2️⃣ 工具(Tool)—— AI 的双手
光有大脑不够,AI 还需要"手"来做事。工具让 AI 能搜索网页、查天气、读文件、算数学题……
🔨 静态工具 —— 固定的技能
静态工具就像瑞士军刀上的固定刀片,出厂就带好了,用的时候直接掏出来。
from langchain_core.tools import tool
# 用 @tool 装饰器定义一个工具(就像教 AI 一项新技能)
@tool
def calculator(expression: str) -> float:
"""计算数学表达式。输入一个数学表达式字符串,返回计算结果。
例如: "2 + 3 * 4" → 14.0
"""
try:
return eval(expression)
except Exception as e:
return f"计算错误: {e}"
@tool
def search_weather(city: str) -> str:
"""查询指定城市的天气。
参数:
city: 城市名称,如"北京"、"上海"
"""
# 这里可以调用真实的天气API
return f"{city}今天晴,温度25°C"
# 把工具绑定到模型上(就像给 AI 装上了手)
llm_with_tools = model.bind_tools([calculator, search_weather])🧩 动态工具 —— 运行时学新技能
动态工具就像你可以随时从 App Store 下载新应用,需要什么就装什么。
from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime
# 定义上下文 —— 运行时传入的信息
@dataclass
class Context:
user_id: str # 当前用户ID
database_url: str # 数据库连接地址
# 动态工具:根据运行时的上下文信息来做不同的事
@tool
def fetch_user_info(runtime: ToolRuntime[Context]) -> str:
"""根据当前登录用户,从数据库查询用户信息。"""
user_id = runtime.context.user_id # 拿到当前用户ID
db_url = runtime.context.database_url # 拿到数据库地址
# 根据 user_id 去数据库查信息(简化示例)
return f"查询到用户 {user_id} 的信息:张三,VIP会员,余额100元"生活类比:你到了一个新城市,下载了当地的地图 App,立刻就能导航。换一个城市,再下另一个 App。
⚠️ 工具错误处理 —— AI 也会犯错
AI 调用工具就像人学骑自行车,偶尔会摔倒。LangChain 提供了三种"安全装备":
from langchain_core.messages import AIMessage, ToolMessage
# ━━━ 方式一:Try/Catch 捕获异常(戴头盔)━━━
# 最简单的方式 —— 出错了就告诉 AI "你搞错了"
def safe_tool_call(tool_args: dict, config) -> str:
try:
return calculator.invoke(tool_args, config=config)
except Exception as e:
# 把错误信息友好地返回给 AI,让它知道自己错了
return f"工具调用失败: {type(e).__name__}: {e}"
# ━━━ 方式二:Fallback 备用方案(带备胎)━━━
# 主工具不行,自动切换到备用工具
chain_with_backup = main_chain.with_fallbacks([backup_chain])
# ━━━ 方式三:自我修正(AI 自己学会纠错)━━━
# 把错误信息反馈给 AI,让它自己想办法重试
# 就像老师批改作业,打回来让学生重做
def self_correct(inputs: dict) -> dict:
exception = inputs.pop("exception")
# 告诉 AI:"上次参数传错了,请修改后重试"
return {
"last_output": [
ToolMessage(content=f"出错啦: {exception}", tool_call_id="xxx"),
]
}生活类比:
- Try/Catch → 摔倒了,爬起来继续走
- Fallback → 这条路堵了,换条路走
- 自我修正 → 摔倒了,想想为什么,下次绕开那个坑
3️⃣ 系统提示(System Prompt)—— 给 AI 的工作手册
系统提示就像你给新员工写的一份工作手册,告诉 AI 它是谁、该做什么、不能做什么。
📋 基础系统提示
from langchain_core.prompts import ChatPromptTemplate, SystemMessagePromptTemplate
# 写一份"工作手册"
system_prompt = """你是一个专业的数学老师。
- 回答问题时,请一步一步地讲解
- 用通俗易懂的语言,举生活中的例子
- 如果学生答错了,不要直接批评,先鼓励再引导
学生问的每一个问题,你都要认真、耐心地回答。"""
# 构建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", system_prompt), # 系统提示(工作手册)
("human", "{user_input}"), # 用户的提问
])
# 把提示和模型串起来
chain = prompt | model
# 调用
response = chain.invoke({"user_input": "什么是勾股定理?"})生活类比:你去肯德基,服务员的工作手册写着"微笑、说欢迎光临、推荐今日特价"——这就是系统提示。
🎭 动态系统提示(结合中间件)
from langchain.agents.middleware import dynamic_prompt, ModelRequest
# 根据不同的场景,动态生成不同的系统提示
@dynamic_prompt
def smart_system_prompt(request: ModelRequest) -> str:
# 检查用户的语言 → 自动切换提示语言
user_message = request.state["messages"][-1].content
if any("\u4e00" <= c <= "\u9fff" for c in user_message):
return "你是一个中文助手,请用中文回答。"
else:
return "You are an English assistant. Please respond in English."4️⃣ 流式传输(Streaming)—— 一边想一边说
普通方式:AI 想完了一整段话,才一次性全发给你。
流式传输:AI 想到一点就说一点,就像真人打字一样,一个字一个字蹦出来。
# ━━━ 方式一:基础流式输出(打字机效果)━━━
for chunk in model.stream("给我讲个故事"):
print(chunk.content, end="", flush=True)
# 输出效果:从...前...有...一...座...山...(一个个字出现)
# ━━━ 方式二:链式流式输出 ━━━
chain = prompt | model
for chunk in chain.stream({"user_input": "写一首诗"}):
print(chunk.content, end="", flush=True)
# ━━━ 方式三:事件流式输出(高级用法,可以看到 AI 的思考过程)━━━
for event in agent.stream_events(
{"messages": [{"role": "user", "content": "3加5等于几?"}]},
version="v2"
):
# 可以看到 AI 正在"思考"、正在"调用工具"、正在"生成回答"
kind = event["event"]
if kind == "on_chat_model_stream":
print(event["data"]["chunk"].content, end="")
elif kind == "on_tool_start":
print(f"\n[AI 正在用工具: {event['name']}]")生活类比:
- 普通输出 → 老师在黑板上写完了一整段话,才让你看
- 流式传输 → 老师一边写,你一边看,像看直播一样
5️⃣ 记忆(Memory)—— AI 的记事本
大模型天生有一个缺点:转头就忘。它不记得上一次说过什么。记忆组件就是给 AI 装一个"记事本"。
🧠 基础记忆 —— 对话历史
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage
# ━━━ 方式一:最简单的记忆(记住所有对话)━━━
memory = ConversationBufferMemory(return_messages=True)
# 第一轮对话
memory.save_context(
{"input": "我叫张三"},
{"output": "你好张三!很高兴认识你。"}
)
# 第二轮对话 —— AI 能记住你叫张三!
memory.save_context(
{"input": "我叫什么名字?"},
{"output": "你叫张三呀!"} # 因为有记忆,所以能回答
)
# ━━━ 方式二:把记忆集成到链中 ━━━
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个友好的助手。"),
MessagesPlaceholder(variable_name="chat_history"), # ← 插入历史记忆
("human", "{input}"),
])
chain = prompt | model
# 第一次对话
chain.invoke({
"input": "我最喜欢的颜色是蓝色",
"chat_history": [] # 第一次没有历史
})
# 第二次对话 —— 把上一次的对话作为记忆传进去
chain.invoke({
"input": "我最喜欢什么颜色?",
"chat_history": [
HumanMessage(content="我最喜欢的颜色是蓝色"),
AIMessage(content="好的,我记住了!你最喜欢蓝色。"),
]
})
# 输出: "你最喜欢的颜色是蓝色!"📝 长期记忆(LangGraph Store)
# LangChain 推荐的新方式 —— 用 LangGraph 的持久化存储
# 就像电脑硬盘,关机了数据也不会丢
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 存储用户偏好
store.put(("users",), "user_001", {"preferences": "喜欢简洁的回答"})
# 以后随时读取
user_data = store.get(("users",), "user_001")
print(user_data.value["preferences"]) # 输出: 喜欢简洁的回答生活类比:
- 没有记忆 → 金鱼记忆(7秒就忘)
- 对话记忆 → 写在便签纸上,聊天时翻看
- 长期记忆 → 写在笔记本上,永久保存
6️⃣ 中间件(Middleware)—— AI 的管家
中间件就像一个贴身管家,站在 AI 和用户之间,负责:
- 检查安全(不能让 AI 说危险的话)
- 控制成本(不能让 AI 无限花钱)
- 日志记录(记下 AI 做了什么)
- 错误重试(失败了自动重试)
⚙️ 内置中间件(开箱即用)
from langchain.agents import create_agent
from langchain.agents.middleware import (
ToolCallLimitMiddleware, # 限制工具调用次数(防止无限循环)
ModelCallLimitMiddleware, # 限制模型调用次数(控制成本)
ToolRetryMiddleware, # 工具失败自动重试
)
agent = create_agent(
model="openai:gpt-4o",
tools=[calculator, search_weather],
middleware=[
ToolCallLimitMiddleware(max_calls=10), # 最多调用10次工具
ModelCallLimitMiddleware(max_calls=5), # 最多调用5次模型
ToolRetryMiddleware(max_retries=3), # 工具失败重试3次
]
)🛠️ 自定义中间件(自己写管家规则)
from langchain.agents.middleware import (
before_model, # 模型调用前触发
after_model, # 模型调用后触发
before_agent, # Agent 开始前触发
after_agent, # Agent 结束后触发
wrap_tool_call, # 包装工具调用(可以做自定义处理)
)
from langchain.agents import AgentState
# ━━━ 管家规则一:每次调用模型前,打印日志 ━━━
@before_model
def log_request(state: AgentState) -> dict | None:
print(f"📝 用户问了: {state['messages'][-1].content[:50]}...")
return None # 返回 None 表示不修改任何东西,只是"看看"
# ━━━ 管家规则二:模型回答后,检查是否包含敏感词 ━━━
@after_model
def safety_check(state: AgentState) -> dict | None:
last_message = state["messages"][-1].content
sensitive_words = ["密码", "银行卡号", "身份证号"]
for word in sensitive_words:
if word in last_message:
# 发现敏感词,替换回答
return {
"messages": [
*state["messages"][:-1],
{"role": "assistant", "content": "抱歉,我不能提供敏感信息。"}
]
}
return None # 没有问题,放行
# ━━━ 管家规则三:包装工具调用,加上耗时统计 ━━━
import time
@wrap_tool_call
def timing_wrapper(tool_call):
start = time.time()
result = yield # 执行原始工具调用
elapsed = time.time() - start
print(f"⏱️ 工具 {tool_call['name']} 耗时: {elapsed:.2f}秒")
return result # 返回原始结果
# 创建 Agent 时注入自定义中间件
agent = create_agent(
model="openai:gpt-4o",
tools=[calculator, search_weather],
middleware=[log_request, safety_check, timing_wrapper]
)生活类比:
@before_model→ 门卫:进出大门前检查一下@after_model→ 质检员:产品出厂前检查质量@wrap_tool_call→ 监工:站在工人旁边盯着干活ToolCallLimitMiddleware→ 限速器:不能开太快ToolRetryMiddleware→ 失败了再来一次
7️⃣ 结构化输出(Structured Output)—— AI 的表格填写能力
默认情况下,AI 的回答是一大段"自然语言"——就像一篇作文。但很多时候你不需要作文,你需要的是填表:姓名、年龄、电话、邮箱……
📋 用 Pydantic 定义输出格式
from pydantic import BaseModel, Field
from typing import Literal
# ━━━ 定义你想要的"表格"格式 ━━━
class PersonalInfo(BaseModel):
"""个人信息表"""
name: str = Field(description="姓名")
age: int = Field(description="年龄", ge=0, le=150)
email: str = Field(description="电子邮箱")
hobbies: list[str] = Field(description="爱好列表")
personality: Literal["内向", "外向", "中间"] = Field(description="性格类型")
# ━━━ 让模型按表格格式输出 ━━━
structured_model = model.with_structured_output(PersonalInfo)
# AI 的输入是一段自然语言,输出是一个结构化的对象!
result = structured_model.invoke("""
我叫王小明,今年25岁。
邮箱是 wangxiaoming@email.com。
我喜欢打篮球、游泳和弹吉他。
我比较外向,喜欢交朋友。
""")
# 直接用对象的方式访问数据!不用解析文本!
print(result.name) # "王小明"
print(result.age) # 25
print(result.email) # "wangxiaoming@email.com"
print(result.hobbies) # ["打篮球", "游泳", "弹吉他"]
print(result.personality) # "外向"
# 还能转成 JSON,方便传给其他系统
print(result.model_dump_json())
# {"name":"王小明","age":25,"email":"wangxiaoming@email.com","hobbies":["打篮球","游泳","弹吉他"],"personality":"外向"}🔄 在 Agent 中使用结构化输出
from langchain.agents import create_agent
from langchain.agents.structured_output import ToolStrategy
# 在 Agent 中使用结构化输出
agent = create_agent(
model="openai:gpt-4o",
tools=[search_weather],
response_format=PersonalInfo # 指定输出格式
)
result = agent.invoke({
"messages": [{"role": "user", "content": "提取信息:张三,30岁,zhang@email.com"}]
})
# 结构化数据在 result["structured_response"] 中
info = result["structured_response"]
print(info.name) # "张三"
print(info.age) # 30生活类比:
- 普通输出 → 你问"今天天气怎么样",AI 回了一大段话
- 结构化输出 → 你给 AI 一张表(日期、温度、天气状况),AI 直接帮你填好
8️⃣ 全部串联 —— 闭环示例 🔄
现在,把上面学的所有零件拼在一起,造一个完整的智能客服机器人!
"""
🎉 完整示例:一个带有记忆、工具、中间件、结构化输出的智能客服
功能:
1. 记住用户信息(记忆)
2. 查询订单和天气(工具)
3. 自动安全检查和日志(中间件)
4. 输出结构化的客服工单(结构化输出)
5. 实时流式显示回答(流式传输)
6. 可以随时切换 AI 模型(动态模型)
"""
from pydantic import BaseModel, Field
from langchain.chat_models import init_chat_model
from langchain.tools import tool, ToolRuntime
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from dataclasses import dataclass
import time
# ═══════════════════════════════════════════
# 1️⃣ 定义上下文(运行时依赖注入)
# ═══════════════════════════════════════════
@dataclass
class AppContext:
user_id: str # 当前用户ID
api_key: str # API密钥(不暴露给AI,只给工具用)
# ═══════════════════════════════════════════
# 2️⃣ 定义工具 —— AI 的"手"
# ═══════════════════════════════════════════
@tool
def query_order(order_id: str, runtime: ToolRuntime[AppContext]) -> str:
"""查询订单状态。
参数:
order_id: 订单编号
"""
user_id = runtime.context.user_id
api_key = runtime.context.api_key
# 用 api_key 调用真实的订单API(这里模拟)
return f"[用户{user_id}] 订单 {order_id} 状态: 已发货,预计明天到达"
@tool
def get_weather(city: str) -> str:
"""查询城市天气。
参数:
city: 城市名称
"""
return f"{city}今天: 晴,25°C,适合出门"
@tool
def calculate(expression: str) -> str:
"""计算数学表达式。
参数:
expression: 数学表达式,如 "100 * 0.8"
"""
try:
return f"计算结果: {eval(expression)}"
except Exception as e:
return f"计算错误,请检查表达式是否正确"
# ═══════════════════════════════════════════
# 3️⃣ 定义中间件 —— AI 的"管家"
# ═══════════════════════════════════════════
from langchain.agents.middleware import before_model, after_model
from langchain.agents import AgentState
@before_model
def log_request(state: AgentState) -> dict | None:
"""每次模型调用前,记录日志"""
msg = state["messages"][-1].content
print(f"\n📋 [日志] 用户输入: {msg[:80]}")
return None
@after_model
def add_cost_warning(state: AgentState) -> dict | None:
"""模型回答后,检查是否包含敏感操作"""
content = state["messages"][-1].content
if "退款" in content or "取消订单" in content:
print("⚠️ [安全] 检测到敏感操作,已通知人工客服")
return None
# ═══════════════════════════════════════════
# 4️⃣ 定义系统提示 —— AI 的"工作手册"
# ═══════════════════════════════════════════
SYSTEM_PROMPT = """你是一个专业的电商客服助手。
- 态度友好、回复及时
- 能查询订单状态、天气信息、进行计算
- 如果用户要退款或取消订单,建议联系人工客服
- 用简洁清晰的语言回答"""
# ═══════════════════════════════════════════
# 5️⃣ 定义结构化输出 —— 客服工单格式
# ═══════════════════════════════════════════
class CustomerTicket(BaseModel):
"""客服工单"""
user_query: str = Field(description="用户的原始问题")
intent: str = Field(description="意图分类: 查订单/查天气/计算/退款/其他")
resolution: str = Field(description="AI 的回答/解决方案")
needs_human: bool = Field(description="是否需要转人工")
# ═══════════════════════════════════════════
# 6️⃣ 组装一切 —— 闭环!
# ═══════════════════════════════════════════
from langchain.agents import create_agent
# 创建完整的 Agent
agent = create_agent(
model="openai:gpt-4o", # 7️⃣ 模型(可以换成动态模型)
tools=[query_order, get_weather, calculate], # 工具
prompt=SYSTEM_PROMPT, # 系统提示
middleware=[log_request, add_cost_warning], # 中间件
response_format=CustomerTicket, # 结构化输出
context_schema=AppContext, # 运行时上下文
)
# ━━━ 调用:传入运行时上下文 ━━━
print("=" * 50)
print("🤖 智能客服已上线!")
print("=" * 50)
# 流式调用 —— 实时显示回答
for chunk in agent.stream(
{
"messages": [{"role": "user", "content": "帮我查一下订单 ORD-2024-001 的状态"}]
},
context=AppContext(user_id="张三", api_key="sk-secret-xxx"), # 注入上下文
):
# 流式显示 AI 的回答过程
if "messages" in chunk:
last = chunk["messages"][-1]
if hasattr(last, "content") and last.content:
print(last.content, end="", flush=True)
# 获取结构化输出(客服工单)
result = agent.invoke(
{
"messages": [{"role": "user", "content": "我要退款!订单 ORD-2024-001"}]
},
context=AppContext(user_id="张三", api_key="sk-secret-xxx"),
)
ticket = result.get("structured_response")
if ticket:
print("\n" + "=" * 50)
print("📋 客服工单(结构化输出):")
print(f" 用户问题: {ticket.user_query}")
print(f" 意图分类: {ticket.intent}")
print(f" 解决方案: {ticket.resolution}")
print(f" 转人工: {'是 ⚠️' if ticket.needs_human else '否'}")
print("=" * 50)🗺️ 完整流程图
用户提问:"查一下我的订单 ORD-001"
│
▼
┌───────────────────┐
│ 📋 系统提示注入 │ ← "你是专业客服,态度友好..."
│ 💾 记忆加载 │ ← "这个用户之前问过天气"
│ ⚙️ 中间件: before │ ← 打印日志: "用户输入: 查一下..."
└───────┬───────────┘
│
▼
┌───────────────────┐
│ 🧠 模型思考 │ ← AI 分析: 用户想查订单
│ 🔧 调用工具 │ ← query_order("ORD-001", runtime)
│ └── 💉 注入上下文 │ ← runtime.context.user_id = "张三"
└───────┬───────────┘
│
▼
┌───────────────────┐
│ ⚙️ 中间件: after │ ← 安全检查: 没有"退款"关键词,放行
│ 💾 保存记忆 │ ← 记住这次对话,下次能用
└───────┬───────────┘
│
▼
┌───────────────────┐
│ 📊 结构化输出 │ ← 转成 CustomerTicket 对象
│ 📡 流式返回 │ ← 用户实时看到回答
└───────────────────┘
│
▼
用户收到回复 ✓📚 总结速查表
| 组件 | 一句话说明 | 生活类比 |
|---|---|---|
| 静态模型 | 固定用一个大模型 | 买了一台手机,固定用 |
| 动态模型 | 运行时切换模型 | 插 SIM 卡,随时换运营商 |
| 静态工具 | 预定义好的工具 | 瑞士军刀的固定刀片 |
| 动态工具 | 运行时按需注入的工具 | App Store 随时下载 |
| 工具错误处理 | Try/Catch、Fallback、自我修正 | 摔倒了爬起来 / 换路 / 学会避坑 |
| 系统提示 | 告诉 AI 它的角色和规则 | 员工工作手册 |
| 流式传输 | 一边生成一边输出 | 直播,不是录播 |
| 记忆 | 让 AI 记住上下文 | 从金鱼变笔记本 |
| 中间件 | 在 AI 执行流程中插入自定义逻辑 | 管家 / 门卫 / 监工 |
| 结构化输出 | 让 AI 按指定格式返回数据 | 填表而不是写作文 |
📝 笔记日期:2026-03-31
📌 参考文档:LangChain 官方文档 · LangChain 中文文档
💡 提示:LangChain 更新很快,建议关注官方 GitHub 获取最新动态
本文是原创文章,采用 CC BY-NC-ND 4.0 协议,完整转载请注明来自 运维小弟