怪兽和他的朋友们

Monster

0%

从入门到从容——LangChain 实战教程

发表于 分类于
本文字数： 7.6k 阅读时长 ≈ 7 分钟

一、前言

2024 年 LangChain 经历了一次大版本重构（v0.3），把过去两年野蛮生长的 API 全部整理了一遍。到了 2025 年，它的生态已经趋于稳定——不再是一个"套壳封装工具"，而是一套负责任的生产级 LLM 编排框架。

如果你对 LangChain 的印象还停留在"不就是包装了一下 OpenAI SDK 嘛"，那可以更新一下了。2025 年的它已经不止 Chain 那套东西——模型路由、工具调用、记忆管理、Agent 编排、调用链追踪，全给你包圆了。

本文不讲概念堆砌，直接从安装开始，带你完成从 Prompt 模板到多步骤 Agent 的全链路实战。

二、安装与快速体验

2.1 安装

2025 年的 LangChain 已经按功能拆分为独立包，按需安装即可：

1
2
3
4
5
6
7
8
# 核心包 + 常用模型 + 工具包
pip install langchain langchain-openai langchain-community

# 如果要走本地模型（Ollama / llama.cpp）
pip install langchain-ollama

# 可观测性（强烈推荐生产环境加）
pip install langsmith

2.2 最小可用链路

先跑通一个最简链路，确认环境正常：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个精通 {field} 的技术顾问，回答要求简洁、准确。"),
    ("human", "{question}"),
])

chain = prompt | llm
result = chain.invoke({
    "field": "Python 异步编程",
    "question": "asyncio 的 gather 和 TaskGroup 有什么区别？",
})
print(result.content)

这里用 | 运算符连接 Prompt 和 LLM，是 v0.3 之后推荐的方式——所有组件实现了 Runnable 协议，可以像 Unix 管道一样组合。不写 LLMChain 类，不调 .run() 方法。

三、核心组件

3.1 Prompt Templates

动态模板比 f-string 多了一层变量校验和类型提示：

1
2
3
4
5
6
7
8
9
10
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个翻译助手，把以下内容翻译成 {target_lang}。"),
    MessagesPlaceholder(variable_name="history"),  # 对话历史占位
    ("human", "{text}"),
])

# 编译后能自动检查必填变量
compiled = prompt.partial(target_lang="日语")

MessagesPlaceholder 是 2025 年处理多轮对话的标准做法，取代了旧版的 chat_history 字符串拼接。

3.2 Output Parsers

LLM 吐给你的永远是一段字符串。但你真正想要的是 JSON、Pydantic 对象、或者一个规整的字典——Parser 就是干这个的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser

class ReviewAnalysis(BaseModel):
    sentiment: str = Field(description="情感倾向: positive/negative/neutral")
    score: int = Field(description="评分 1-5", ge=1, le=5)
    keywords: list[str] = Field(description="提取的关键词")

parser = PydanticOutputParser(pydantic_object=ReviewAnalysis)

chain = prompt | llm | parser
result: ReviewAnalysis = chain.invoke({"review": "电池续航太差，但屏幕很不错。"})
# result.sentiment -> "negative"
# result.score -> 2
# result.keywords -> ["电池续航", "屏幕"]

Parser 会自动注入 format instructions 到 Prompt 里，保证模型输出能被正确解析。出错时还会触发重试机制。

3.3 模型参数绑定

同一个模型对不同任务可能需要不同的 temperature、max_tokens。用 .bind()：

1
2
3
4
5
6
7
8
# 代码生成要低 temperature
code_llm = llm.bind(temperature=0.1, max_tokens=4096)

# 创意写作要高 temperature
creative_llm = llm.bind(temperature=0.9)

# 结构化输出可以直接绑定 response_format
json_llm = llm.bind(response_format={"type": "json_object"})

不创建多个 LLM 实例，同一个实例通过 bind 派生专用变体，这是 Runnable 协议带来的工程便利。

四、实战场景

4.1 构建 RAG 问答系统

RAG（检索增强生成）算是 LangChain 用得最多的场景了。2025 年写起来长这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain_core.runnables import RunnablePassthrough
from langchain_core.documents import Document

# 1. 加载文档
docs = [
    Document(page_content="LangChain v0.3 于 2024 年 10 月发布，重构了核心 API。"),
    Document(page_content="Runnable 协议让所有组件可以通过 | 运算符组合。"),
    Document(page_content="LangSmith 提供全链路追踪和评估能力。"),
]

# 2. 建立向量库
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 3. 构建 RAG 链路
template = """基于以下上下文回答问题。如果找不到答案，直接说不知道。
上下文：{context}
问题：{question}"""
prompt = ChatPromptTemplate.from_template(template)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
)

# 4. 执行
print(rag_chain.invoke("LangChain v0.3 什么时候发布的？").content)
# -> "2024 年 10 月发布。"

注意这里的 RunnablePassthrough()——它把用户输入原样传递下去，同时 retriever 异步去查向量库，最终在 Prompt 模板里合并。不需要手写 for 循环拼接上下文。

4.2 带工具调用的 Agent

Agent 是 LangChain 最有威力的模式：让 LLM 决定调用什么工具、传入什么参数、然后基于工具返回结果继续推理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

@tool
def get_stock_price(symbol: str) -> float:
    """查询股票当前价格"""
    # 这里是伪代码，实际对接行情 API
    import random
    return round(random.uniform(10, 500), 2)

@tool
def calculate(expression: str) -> float:
    """执行数学计算"""
    return eval(expression)  # 注意：生产环境要用安全的 math parser

tools = [get_stock_price, calculate]

agent = create_tool_calling_agent(
    llm=ChatOpenAI(model="gpt-4o", temperature=0),
    tools=tools,
    prompt=ChatPromptTemplate.from_messages([
        ("system", "你是一个金融助手，使用工具回答用户问题。"),
        ("placeholder", "{chat_history}"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ]),
)

agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({
    "input": "计算苹果公司当前市值的 1.5%，已知苹果总股本 155 亿股"
})

执行时 LLM 的思考过程：

  1. 识别到需要股价 → 调用 get_stock_price("AAPL")
  2. 拿到股价 → 调用 calculate("155 * current_price * 0.015")
  3. 组装最终答案返回

agent_scratchpad 是中间思考过程的容器，AgentExecutor 会自动写入每次工具调用的输入和输出。不需要手动维护状态。

4.3 多步骤流水线

复杂的业务逻辑需要多个 LLM 调用串联，每个步骤的输出作为下一步的输入：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
from langchain_core.runnables import RunnableLambda

# 步骤 1：分类客户意图
intent_chain = classify_prompt | llm.bind(temperature=0) | intent_parser

# 步骤 2：根据意图生成回答
def route_by_intent(intent: Intent):
    if intent.type == "complaint":
        return complaint_chain
    elif intent.type == "inquiry":
        return inquiry_chain
    else:
        return default_chain

# 步骤 3：质量检查
def quality_check(response: str) -> str:
    if len(response) < 10:
        return "【需重写】回答过短，请补充细节。"
    return response

full_pipeline = (
    intent_chain
    | RunnableLambda(route_by_intent)
    | RunnableLambda(quality_check)
)

result = full_pipeline.invoke({"message": "你们的产品坏了怎么办？"})

用 RunnableLambda 把纯 Python 函数包成 Runnable，整个流水线保持 | 链式调用，类型检查、错误重试、日志追踪全部自动继承。

五、生产环境要点

5.1 可观测性

LangSmith 是 LangChain 生态里的标准追踪方案。加一行配置就能看到每次调用的完整链路：

1
2
3
4
5
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "ls_xxx"

# 后续所有 chain.invoke / agent_executor.invoke 自动上报

LangSmith 的 Web UI 里能看到：

指标 作用
延迟分布 哪个步骤最慢，瓶颈在 LLM 还是工具
Token 消耗 每次调用的 prompt / completion token 数
失败率 解析失败、工具超时、模型拒绝回答
输入输出采样 回溯具体某次异常对话

5.2 缓存

对重复查询场景，LLM 返回的内容通常可以缓存：

1
2
3
4
5
6
7
8
9
10
11
12
from langchain_core.cache import InMemoryCache, SQLiteCache

# 开发环境用内存缓存
llm.cache = InMemoryCache()

# 生产环境持久化到 SQLite
llm.cache = SQLiteCache(database_path=".llm_cache.db")

# 首次调用写入缓存
result1 = llm.invoke("Python 中如何合并两个字典？")
# 第二次相同输入直接从缓存返回，零延迟
result2 = llm.invoke("Python 中如何合并两个字典？")

缓存命中基于模型 + prompt + temperature 的完整签名，同一个问题问 temperature=0 和 temperature=0.7 不会混用。

5.3 错误处理与重试

LLM 调用可能因 API 限流、网络抖动、输出解析失败而报错。Runnable 内置了重试机制：

1
2
3
4
5
6
from langchain_core.runnables import RunnableConfig

chain = prompt | llm.with_retry(
    retry_if_exception=lambda e: isinstance(e, (TimeoutError, ConnectionError)),
    stop_after_attempt=3,
)

对于 Pydantic 解析失败的情况，LangChain 会自动触发"格式化重试"——把错误信息塞回给 LLM，让 LLM 修正输出。不需要写 try/except 循环。

六、总结

组件 用途 关键方法 / 运算符
ChatPromptTemplate 多轮对话模板 from_messages()
OutputParser 结构化输出 PydanticOutputParser
RunnablePassthrough 透传输入 \| 链式组合
Retriever 向量检索 as_retriever()
Tool 工具函数定义 @tool 装饰器
Agent / AgentExecutor 自主决策执行 create_tool_calling_agent()
RunnableLambda 函数包装为 Runnable \| 链式组合
Cache LLM 输出缓存 InMemoryCache / SQLiteCache

LangChain 的设计哲学总结成一句话：让 LLM 调用像 Unix 管道一样可组合、可追踪、可测试。

到了 2025 年，AI 应用开发早过了"调通 OpenAI SDK 就收工"的阶段。prompt 版本怎么管、工具调用怎么容错、多步骤链路怎么追踪——这些问题不是加几个 if else 能解决的。LangChain 给你铺好了路，但用好它还是得靠你自己把业务逻辑拆明白。框架不替代思考，它只是让你的思考能体面落地。

打开终端，pip install langchain，跑通你的第一个 chain。从那一刻开始，你写的就不是"脚本"了，是 AI 应用。