LangChain 的学习与应用（一）：模型调用与提示词基础

如果你已经会用某一家大模型的 API，却在换模型、做流式输出、批量跑任务时被各家 SDK 的差异折磨过，那么 LangChain 大概率就是你要找的东西。这是本系列的第一篇，从“为什么需要它”讲起，带你走完初始化模型、四种消息写法、同步/异步、流式输出、批量调用和提示词模板这条主线；第二篇会继续讲输出解析与链式编排。

一、为什么需要 LangChain

直接用官方 SDK 调大模型当然可行，OpenAI、Anthropic、Google 都提供了自己的 Python 库。但只要项目稍微认真一点，你很快会撞上两类问题：

第一类：接口不统一。 每家厂商的 SDK 参数名、返回结构、流式协议都不一样。今天用 GPT，明天想切换到 DeepSeek 或 Gemini 对比效果，业务代码就得跟着改一遍。模型能力迭代很快，“随时可以换模型”应该是架构能力，而不是重构任务。

第二类：模型本身有短板。 裸模型不会联网搜索、不会调用外部工具、不会记住上一轮对话。这些能力都需要在模型之外用工程手段补齐——提示词拼装、工具调用循环、对话历史管理……每一件事都不难，但每一件事都要自己写，就很难了。

LangChain 解决的正是这两件事：

统一异构模型的 API——一套调用语法适配所有主流模型厂商，切换模型基本只改一个字符串；
提供任务编排能力——把提示词、模型、工具、记忆、知识库串成一条可维护的链路。

从学习路线的角度，LangChain 通常被拆成四大模块，也正好是循序渐进的四个阶段：

模块	解决什么问题
Model I/O	构建输入（提示词模板）、调用模型、解析输出
Chains	把多个步骤编排成链，实现多步任务
RAG	检索增强生成，让模型基于你的私有文档回答问题
Agents	让模型自主决策调用哪些工具、如何完成任务

本文聚焦在 Model I/O——它是一切的地基，后面三个模块都建立在“会调模型”这件事之上。

二、准备工作：安装与密钥管理

安装很简单，按你要用的厂商装对应的扩展即可：

pip install -U "langchain[openai]"    # OpenAI 及兼容 OpenAI 协议的模型
pip install -U "langchain[anthropic]" # Anthropic Claude
pip install -U "langchain[google-genai]" # Google Gemini

API Key 这类敏感信息不要写死在代码里。推荐的做法是放进环境变量，有两种常见方案：

系统环境变量：适合生产环境，配置后记得重启 IDE 才能生效；
项目内的 .env 文件：适合开发环境，配合 python-dotenv 读取。注意 .env 要加入 .gitignore，另外维护一份脱敏的 .env.example 提交到仓库，方便协作者知道需要配置哪些变量。

from dotenv import load_dotenv

# 读取 .env 文件中的 OPENAI_API_KEY、OPENAI_BASE_URL 等配置
# 默认系统环境变量优先；override=True 则让 .env 文件优先
load_dotenv(override=True)

一个容易踩的坑：当同名变量既存在于系统环境又存在于 .env 中时，load_dotenv() 默认不会覆盖系统变量，需要显式传 override=True 才以文件为准。

三、第一次调用：init_chat_model

LangChain 1.x 推荐用 init_chat_model 这个统一入口来初始化任何厂商的聊天模型。模型的指定方式有两种写法，效果等价：

from langchain.chat_models import init_chat_model

# 写法1：用 "厂商:模型名" 一个字符串搞定
model = init_chat_model("openai:gpt-4o-mini")

# 写法2：模型名和厂商分开传
model = init_chat_model(model="gpt-4o-mini", model_provider="openai")

response = model.invoke("用一句话介绍一下 LangChain")
print(response.content)

这就是“统一接口”的含义：把 "openai:gpt-4o-mini" 换成 "anthropic:claude-sonnet-4-5" 或 "google_genai:gemini-2.5-flash"，后面的代码一行都不用动。

除了统一入口，每个厂商还有对应的模型类，可以直接实例化。比如 OpenAI 对应 ChatOpenAI（来自 langchain-openai 包）：

from langchain_openai import ChatOpenAI

# 写法3：直接实例化厂商模型类
model = ChatOpenAI(model="gpt-4o-mini")

两条路线怎么选：厂商已经确定、或需要用到厂商特有参数时，直接用模型类更直观；想保留随时换厂商的灵活性，就用 init_chat_model。另外一个实用技巧——很多国产模型（DeepSeek、Qwen 等）都兼容 OpenAI 协议，用 ChatOpenAI 配上对应的 base_url 和 api_key 就能直接调用。

init_chat_model 还接受一批通用参数，用来控制模型行为：

model = init_chat_model(
    "openai:gpt-4o-mini",
    temperature=0.7,   # 随机性：越高越有创造性，越低越稳定
    max_tokens=1000,   # 限制回复长度
    timeout=30,        # 超时时间（秒）
    max_retries=6,     # 失败自动重试次数（默认 6，指数退避）
)

值得一提的是 max_retries：LangChain 会对网络错误、限流（429）、服务端错误（5xx）自动重试，而 401、404 这类客户端错误不会重试。网络环境不稳定时可以适当调大。

四、消息的四种写法

invoke() 的输入不只可以是一个字符串。实际对话中往往需要区分角色——system（系统设定）、human/user（用户）、ai/assistant（模型回复）等。LangChain 接受四种等价的消息写法：

from langchain_core.messages import SystemMessage, HumanMessage

# 1. 纯字符串：最简单，等价于一条 human 消息
model.invoke("你好")

# 2. 消息对象列表：类型明确，IDE 提示友好
model.invoke([
    SystemMessage(content="你是一个专业翻译"),
    HumanMessage(content="把'早上好'翻译成英文"),
])

# 3. 元组列表：(角色, 内容)，写起来最省事
model.invoke([
    ("system", "你是一个专业翻译"),
    ("human", "把'早上好'翻译成英文"),
])

# 4. 字典列表：和 OpenAI 原生格式一致，方便迁移旧代码
model.invoke([
    {"role": "system", "content": "你是一个专业翻译"},
    {"role": "user", "content": "把'早上好'翻译成英文"},
])

四种写法喂给模型的内容完全相同，选哪种看场景：快速实验用元组，正式项目用消息对象，从 OpenAI SDK 迁移过来的代码用字典最顺手。

五、同步与异步调用

invoke() 是同步调用：发出请求后线程原地等待，直到模型返回完整结果。写脚本、做实验完全够用。但在 Web 服务这类高并发场景里，模型一次推理动辄好几秒，同步等待会白白占住线程。这时候就需要异步版本 ainvoke()：

import asyncio
from langchain.chat_models import init_chat_model

model = init_chat_model("openai:gpt-4o-mini")

async def main():
    # await 期间线程被释放，可以去处理其他任务
    response = await model.ainvoke("你好")
    return response

result = asyncio.run(main())
print(result.content)

LangChain 的命名规律很好记：同步方法前面加个 a 就是异步版本——invoke/ainvoke、stream/astream、batch/abatch。异步真正的威力在并发：

async def concurrent_calls():
    # 三个请求同时发出，总耗时约等于最慢的那一个
    results = await asyncio.gather(
        model.ainvoke("介绍一下 Python"),
        model.ainvoke("介绍一下 Java"),
        model.ainvoke("介绍一下 Go"),
    )
    for r in results:
        print(r.content[:50])

一句话总结选择标准：写脚本、单次调用用 invoke；在 FastAPI 等异步框架里、或需要并发多个请求时用 ainvoke。

六、流式输出

用过 ChatGPT 的人都熟悉“打字机效果”。它背后就是流式输出：模型每生成一小段就立刻推给客户端，而不是憋到全部生成完才返回。对长回复来说，这是用户体验的分水岭——没人愿意面对一个转了十秒的加载圈。

把 invoke() 换成 stream() 就行，返回值从单条消息变成一个迭代器：

for chunk in model.stream("写一段 200 字的春天散文"):
    print(chunk.content, end="", flush=True)

invoke() 返回一个完整的 AIMessage，而 stream() 逐个产出 AIMessageChunk。如果既想流式展示、又想在结束后拿到完整消息，chunk 之间支持直接相加聚合：

full = None
for chunk in model.stream("介绍一下流式输出"):
    print(chunk.content, end="", flush=True)
    full = chunk if full is None else full + chunk

# full 此时是一条完整消息，可以存入对话历史继续使用

对应的异步版本是 astream()，配合 async for 使用，适合在异步 Web 框架里做 SSE 推送。

七、批量调用

假设你要给 1000 条用户评论做情感分类。写个 for 循环挨个 invoke()？串行执行，每条等一次网络往返，慢得让人绝望。batch() 就是为这种场景准备的——它在客户端用线程池把请求并行发出去：

questions = [
    [("system", "你是一个大模型讲师"), ("human", "如何学习 LangChain？")],
    [("system", "你是一个大模型讲师"), ("human", "如何学习 LangGraph？")],
    [("system", "你是一个大模型讲师"), ("human", "RAG 是什么？")],
]

responses = model.batch(questions)
for r in responses:
    print(r.content[:80])
    print("-" * 40)

几个实用细节：

控制并发数。 请求太多时并发全开容易触发厂商限流，可以通过 max_concurrency 限制同时在途的请求数：

responses = model.batch(
    questions,
    config={"max_concurrency": 5},  # 最多 5 个请求并行
)

先完成先返回。 batch() 会等所有请求做完才一起返回。如果想每完成一条就处理一条，用 batch_as_completed()，注意结果是乱序到达的（每个结果自带原始索引，需要时可据此还原顺序）：

for idx, response in model.batch_as_completed(questions):
    print(f"第 {idx} 条完成：{response.content[:50]}")

异步版本。 abatch() 与 batch() 行为一致，只是基于协程而非线程池，适合在异步应用中使用。

另外要区分一个概念：这里的 batch() 是客户端并行，请求实时返回、按标准价格计费；OpenAI 等厂商还提供服务端的 Batch API（延迟高但价格减半），那是另一个东西，不要混淆。

八、提示词模板：从写死到复用

前面的例子里提示词都是写死的字符串。实际应用中，提示词往往是“固定框架 + 动态内容”——框架不变，只有用户输入的部分在变。ChatPromptTemplate 让你用占位符定义框架，运行时再填充：

from langchain_core.prompts import ChatPromptTemplate

# 用 {} 定义槽位
template = ChatPromptTemplate.from_messages([
    ("system", "你是一位资深的{role}"),
    ("human", "请帮我解答关于{topic}的问题"),
])

# 填充槽位，生成真正发给模型的消息列表
messages = template.invoke({"role": "Python 讲师", "topic": "装饰器"})

response = model.invoke(messages)
print(response.content)

模板的价值不只是字符串拼接的语法糖：它把提示词从业务逻辑中剥离出来，变成可以单独维护、测试、版本化的资产。当你的应用有几十个提示词时，这种结构化管理的收益会非常明显。它也是下一阶段学习 Chains 的入口——模板、模型、输出解析器正是链的基本组成单元。

九、小结与学习建议

回顾一下这条主线：

为什么用 LangChain：统一异构模型的接口 + 补齐裸模型缺失的工程能力；
init_chat_model：一个入口初始化任意厂商模型，换模型只改一个字符串；
四种消息写法：字符串、消息对象、元组、字典，按场景选用；
invoke / ainvoke：脚本用同步，高并发服务用异步；
stream / astream：长回复必备的打字机体验，chunk 可相加聚合；
batch / abatch：批量任务并行处理，记得用 max_concurrency 防限流；
ChatPromptTemplate：把提示词变成可复用、可维护的模板。

这些内容对应 LangChain 四大模块中的 Model I/O。掌握之后，建议按 Chains → RAG → Agents 的顺序继续推进：先学会把步骤串成链，再学会让模型基于私有知识回答问题，最后学会让模型自主使用工具。每一步都以前一步为基础，不建议跳跃。本系列的第二篇《输出解析与链式编排》会接着讲：如何让模型稳定输出 JSON 和结构化对象，以及如何用一个管道符把提示词、模型、解析器串成链。

最后一个建议：LangChain 迭代很快，1.x 版本和网上大量旧教程（0.x 时代）的 API 差异不小，遇到不一致时以官方文档为准。动手把本文的每个代码片段跑一遍，比读十篇文章都有用。