LangChain 模型调用：从 Model I/O 到流式、异步与多模态

调用一次大语言模型并不困难，真正困难的是把模型稳定地接入应用：输入需要带有明确的角色和上下文，输出需要保留内容、工具调用与 Token 用量；系统还要支持流式响应、批处理、异步并发、超时重试、速率限制以及模型切换。

LangChain 的 Model I/O 为这些需求提供了一组相对统一的抽象。开发者可以使用相同的调用接口连接不同模型，并把模型继续接入提示词模板、结构化输出、RAG 和 Agent 工作流。

本文将系统介绍 LangChain 1.x 中的模型类型、消息协议和调用方式，并讨论多模态、本地模型、用量追踪与提示词缓存等工程问题。

LangChain 与模型提供商的 API 都在持续更新。示例中的模型名称仅用于展示调用方式，实际项目应选择账户可用的模型、锁定依赖版本，并查阅对应版本的官方文档。

一、理解 Model I/O

Model I/O 是 LangChain 中负责模型输入与输出的基础层，可以拆成三个环节：

业务数据
   ↓
Prompts：将变量、系统指令和上下文组织成提示
   ↓
Models：调用聊天模型或其他模型
   ↓
Output Parsers / Structured Output：解析并验证模型结果
   ↓
业务对象

• Prompts 解决“如何构造模型输入”；
• Models 解决“如何以统一方式调用模型”；
• Output Parsers 解决“如何把模型输出转换为程序可处理的数据”。

本文聚焦中间的 Models 层。提示词模板和输出解析虽然不是模型本身，却共同决定了一次模型调用的输入契约与输出契约。

二、LangChain 中的三类模型

LangChain 语境中的“模型”并不都用于生成回答。需要区分以下三类接口：

类型	典型输入	典型输出	主要用途
Chat Models	字符串或消息序列	AIMessage	对话、文本生成、工具调用、多模态任务
LLMs	字符串	字符串	兼容传统文本补全模型
Embeddings	文本或文本列表	浮点数向量	语义检索、聚类和 RAG

现代生成式模型通常通过 Chat Model 接口使用。Embeddings 不生成自然语言，而是将文本映射到向量空间，两者不能互换。传统 LLM 接口仍存在于生态中，但新项目一般优先使用 Chat Model。

三、为什么使用统一模型接口

直接使用模型提供商的 SDK 完全合理，尤其适合只使用单一平台、需要最新专有能力或希望直接控制底层请求的项目。LangChain 的价值主要体现在跨组件组合和接口一致性上。

不同提供商的客户端初始化、消息格式、结果对象和流式协议可能不同。LangChain 将常见能力归一为以下接口：

response = model.invoke(input_data)
response = await model.ainvoke(input_data)
responses = model.batch(inputs)

for chunk in model.stream(input_data):
    ...

统一接口可以减少应用层适配代码，但不意味着模型可以零成本替换。不同模型在上下文窗口、工具调用、结构化输出、多模态内容、参数范围和安全策略上仍然存在差异。每次切换都应重新进行功能与质量测试。

四、准备环境并初始化模型

以 OpenAI 集成为例，使用 uv 安装依赖：

uv add langchain langchain-openai python-dotenv

在项目根目录创建 .env：

OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://api.openai.com/v1

.env 包含敏感凭据，必须加入 .gitignore。生产环境应使用部署平台提供的 Secret 管理能力，而不是把密钥写进代码、镜像或配置仓库。

1. 使用提供商专用类

模型提供商固定时，专用类通常拥有更清晰的参数提示和更完整的特性支持：

from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0,
    timeout=30,
    max_retries=2,
)

response = model.invoke("请用一句话解释 LangChain。")
print(response.content)

ChatOpenAI 会按照集成约定读取 OPENAI_API_KEY 等环境变量。如果使用兼容 OpenAI 协议的第三方端点，应显式确认其鉴权、参数、流式输出、工具调用和结构化输出兼容程度；“协议兼容”通常不等于所有能力完全一致。

Anthropic、Google 和 Ollama 等平台通常使用各自的集成包和专用类：

uv add langchain-anthropic
uv add langchain-google-genai
uv add langchain-ollama

2. 使用 init_chat_model

需要由配置或用户选择动态决定提供商时，可以使用 init_chat_model()：

from langchain.chat_models import init_chat_model

model = init_chat_model(
    model="gpt-4.1-mini",
    model_provider="openai",
    temperature=0,
)

response = model.invoke("介绍一下 Model I/O。")

init_chat_model() 统一了初始化入口，适合 A/B 测试、多租户配置和运行时选型。若项目只连接一个确定的提供商，直接使用 ChatOpenAI、ChatAnthropic 等专用类通常更直观。

五、消息是聊天模型的基础协议

1. 四种核心消息类型

LangChain 使用标准消息对象表达角色和内容：

消息类型	作用
SystemMessage	定义模型角色、全局规则和行为边界
HumanMessage	表示用户输入
AIMessage	表示模型输出，也可用于构造历史对话
ToolMessage	将工具执行结果返回给模型

基本调用如下：

from langchain_core.messages import HumanMessage, SystemMessage

messages = [
    SystemMessage(content="你是一名严谨的 Python 编程助手。"),
    HumanMessage(content="什么是装饰器？"),
]

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

系统消息只是提供给模型的指令，不应被视为真正的安全边界。权限检查、工具白名单和数据访问控制必须由应用代码执行。

2. 三种常用输入方式

聊天模型通常接受字符串、消息对象列表或角色元组：

# 单轮问答
model.invoke("什么是 LangChain？")

# 明确的消息对象
model.invoke(
    [
        SystemMessage(content="你是翻译助手。"),
        HumanMessage(content="将“你好”翻译为英文。"),
    ]
)

# 简洁的角色元组
model.invoke(
    [
        ("system", "你是翻译助手。"),
        ("human", "将“你好”翻译为英文。"),
    ]
)

直接传字符串适合快速测试；消息列表适合角色设定和多轮上下文；动态提示词则更适合交给 ChatPromptTemplate 管理，而不是在业务代码中手动格式化字符串。

3. 对话历史不等于持久化记忆

可以把先前的用户消息和模型回复重新放入消息列表：

from langchain_core.messages import AIMessage, HumanMessage

conversation = [
    HumanMessage(content="我正在学习 LangChain。"),
    AIMessage(content="好的，我们可以从 Model I/O 开始。"),
    HumanMessage(content="刚才建议我从哪里开始？"),
]

response = model.invoke(conversation)

这只是本次请求携带了历史消息，模型本身并没有永久“记住”对话。跨请求的会话状态仍需由应用保存、裁剪并重新注入。历史越长，输入 Token、延迟和费用通常越高，因此生产系统还需要摘要、窗口裁剪或检索式记忆策略。

六、理解 AIMessage 返回值

invoke() 返回的通常不是普通字符串，而是 AIMessage：

response = model.invoke("介绍一下 Runnable。")

print(response.content)
print(response.response_metadata)
print(response.usage_metadata)
print(response.tool_calls)

常见字段包括：

• content：文本或多模态内容块；
• response_metadata：模型名称、结束原因和提供商响应信息；
• usage_metadata：输入、输出和总 Token 用量，是否可用取决于集成和响应模式；
• tool_calls：模型请求执行的工具调用；
• id：本次消息或响应的标识。

只读取 .content 适合简单文本展示；构建生产应用时还应检查结束原因、工具调用和用量元数据。不要假设所有提供商返回完全相同的元数据字段。

七、常用模型参数

model = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0,
    timeout=30,
    max_retries=2,
    max_tokens=800,
)

1. model

model 指定服务端模型名称。模型名称不是跨平台标准，应使用提供商实际支持的标识，并避免把预览版名称长期写死在业务逻辑中。

2. temperature

temperature 影响采样随机性。较低值通常更稳定，较高值通常增加输出多样性，但它不是“事实正确率”开关，也不能保证设置为 0 后结果绝对一致。

可按任务性质选择：

任务	常见倾向
信息抽取、分类、代码修改	使用较低随机性
摘要、一般问答	使用低至中等随机性
创意写作、方案发散	在可接受范围内提高随机性

具体取值范围和行为由模型提供商定义，部分推理模型可能忽略或限制该参数。

3. max_tokens

该参数限制最大输出 Token 数，具体参数名可能因集成或 API 版本而不同。设置过小可能导致回答被截断，但回答提前结束也可能是模型判断任务已完成或触发其他停止条件，因此应同时检查响应元数据中的结束原因。

4. timeout 与 max_retries

timeout 限制单次网络请求的等待时间，max_retries 控制集成层对部分暂时性错误的重试。重试应配合指数退避和并发限制使用；权限错误、无效参数和确定性业务错误通常不应反复重试。

八、Token：模型实际处理的计量单位

Token 不是固定数量的汉字、字母或单词，而是由当前模型的分词器切分出的文本片段。同一句话在不同模型或不同分词器下，Token 数可能不同。

一次调用的主要用量通常包括：

总 Token = 输入 Token + 输出 Token

输入不仅包含用户问题，还可能包括系统提示词、历史消息、检索上下文、工具定义和多模态描述。控制成本不能只限制回答长度，还要管理传入模型的完整上下文。

估算 Token 时应使用模型提供商推荐的分词工具；计费和审计应以 API 返回的实际用量为准，而不是使用“一个 Token 约等于几个字”的经验公式。

九、六种常用调用方式

1. invoke：同步单次调用

response = model.invoke("什么是 Model I/O？")
print(response.content)

适合脚本、低并发任务和简单原型。调用期间当前线程会等待网络响应。

2. ainvoke：异步单次调用

import asyncio


async def main() -> None:
    response = await model.ainvoke("什么是异步调用？")
    print(response.content)


asyncio.run(main())

异步的价值是等待网络响应时让出事件循环，而不是让一次模型推理本身变快。Jupyter 通常已有事件循环，可以直接执行 await model.ainvoke(...)；普通 Python 脚本通常使用 asyncio.run()。

3. 并发执行多个异步调用

连续写多个 await 仍然是顺序等待。要让相互独立的请求并发执行，可以使用 asyncio.gather()：

import asyncio


async def ask_many(questions: list[str]):
    tasks = [model.ainvoke(question) for question in questions]
    return await asyncio.gather(*tasks)

并发数量不能无限增加。实际系统必须考虑服务商速率限制、本地连接池、超时、费用和失败隔离，通常还需要使用信号量或任务队列限制并发。

4. stream：同步流式调用

full_response = None

for chunk in model.stream("请解释流式响应的工作方式。"):
    full_response = chunk if full_response is None else full_response + chunk
    print(chunk.content, end="", flush=True)

print("\n完整结果：", full_response.content)

流式响应可以缩短用户感知到的首字延迟，但不一定减少总推理时间或 Token 费用。应用还要处理客户端断开、流式错误、内容审核以及消息块合并。

5. astream：异步流式调用

async def stream_answer() -> None:
    async for chunk in model.astream("生成一段简短说明。"):
        print(chunk.content, end="", flush=True)

异步 Web 框架通常使用 astream() 将模型数据块转发给客户端。具体传输层可以使用 Server-Sent Events、WebSocket 或流式 HTTP 响应。

6. batch 与 abatch：批量处理

questions = [
    "什么是 Chat Model？",
    "什么是 Embedding？",
    "什么是 Output Parser？",
]

responses = model.batch(questions)

for question, response in zip(questions, responses):
    print(question, response.content)

异步环境可以使用：

responses = await model.abatch(questions)

批处理适合多个相互独立的输入。它是否使用提供商原生批量 API、并发请求还是其他策略，取决于具体 Runnable 和模型集成，不能仅从方法名推断底层实现。

十、运行时配置与可观测性

config 用于传递 LangChain 运行时信息，而不是模型业务输入：

response = model.invoke(
    "请生成一段摘要。",
    config={
        "tags": ["summary", "production"],
        "metadata": {"request_type": "document-summary"},
    },
)

标签和元数据可以被回调系统或 LangSmith 等追踪平台采集，用于调试、监控和分析。应避免把 API Key、完整个人信息或其他敏感数据放入可观测性元数据，因为这些信息可能进入日志和外部追踪系统。

对于生产模型调用，建议至少记录：

• 请求类型与模型版本；
• 延迟、错误类型和重试次数；
• 输入、输出和缓存 Token 用量；
• 流程追踪 ID；
• 是否发生工具调用或回退。

原始提示词和模型输出是否记录，应根据隐私、合规和数据保留政策决定。

十一、接入本地 Ollama 模型

Ollama 可以在本地运行部分开源模型，适合离线实验、隐私敏感原型和本地开发。安装 Ollama 并下载模型后，添加 LangChain 集成：

uv add langchain-ollama

from langchain_ollama import ChatOllama

local_model = ChatOllama(
    model="your-local-model",
    base_url="http://localhost:11434",
)

response = local_model.invoke("介绍一下本地模型的优势。")
print(response.content)

“本地运行”不代表没有成本。实际性能取决于模型规模、量化方式、内存、显存和推理后端；模型能力也可能与云端模型不同。上线前需要评估吞吐量、并发、硬件利用率和模型许可证。

十二、多模态输入

支持视觉输入的模型可以接收文本和图片内容块。图片可以使用可访问的 URL，也可以编码为 Data URL：

import base64

from langchain_core.messages import HumanMessage


def image_to_data_url(path: str, media_type: str = "image/jpeg") -> str:
    with open(path, "rb") as image_file:
        encoded = base64.b64encode(image_file.read()).decode("ascii")
    return f"data:{media_type};base64,{encoded}"


message = HumanMessage(
    content=[
        {"type": "text", "text": "请描述图片中的主要内容。"},
        {
            "type": "image_url",
            "image_url": {"url": image_to_data_url("image.jpg")},
        },
    ]
)

response = vision_model.invoke([message])

内容块 Schema 会随模型集成和提供商 API 演进，使用前应查看对应集成文档。Base64 只是一种传输编码，不是加密方式；编码后的图片仍然属于敏感数据，并且体积通常会增加。上传前还应限制文件类型和大小，移除不必要的元数据，并执行访问控制。

十三、速率限制与用量追踪

1. 客户端速率限制

LangChain 提供内存速率限制器，用于控制当前进程的请求节奏：

from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain_openai import ChatOpenAI

rate_limiter = InMemoryRateLimiter(
    requests_per_second=1,
    check_every_n_seconds=0.1,
    max_bucket_size=2,
)

limited_model = ChatOpenAI(
    model="gpt-4.1-mini",
    rate_limiter=rate_limiter,
)

内存限流器只约束当前进程，无法协调多个服务实例。分布式系统需要使用共享限流基础设施，并同时处理提供商返回的速率限制响应。

2. 聚合 Token 用量

可以在一段调用范围内聚合模型返回的用量元数据：

from langchain_core.callbacks import get_usage_metadata_callback

with get_usage_metadata_callback() as usage:
    model.invoke("什么是输入 Token？")
    model.invoke("什么是输出 Token？")
    print(usage.usage_metadata)

用量是否完整取决于提供商、模型和流式配置。成本核算应结合提供商账单，而不是只依赖客户端估算。

十四、区分提示词缓存与结果缓存

这两种“缓存”解决的是不同问题：

对比项	提示词缓存	LLM 结果缓存
缓存位置	通常位于模型提供商服务端	应用内存、磁盘或 Redis 等存储
缓存内容	重复提示前缀的内部计算状态	完整的输入到输出映射
是否仍调用模型 API	是	命中时通常不调用
主要价值	降低重复前缀的延迟或费用	对相同请求直接复用结果

提示词缓存的启用条件、最小长度、有效期和计费方式由模型提供商决定，而且可能随时调整。为了提高命中率，稳定的系统指令应放在提示前部，动态内容放在后部，但不能为了缓存而破坏提示词语义。

结果缓存更可控，但只适合允许复用答案的确定性或低时效任务。包含用户身份、实时数据、权限上下文或随机创作要求的请求，必须谨慎设计缓存键和隔离范围。

十五、完整示例：可流式调用的技术问答模型

下面的示例把环境配置、消息协议、运行时元数据和流式输出组合起来：

import os

from dotenv import load_dotenv
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

load_dotenv()

model = ChatOpenAI(
    model="gpt-4.1-mini",
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1"),
    temperature=0,
    timeout=30,
    max_retries=2,
)

messages = [
    SystemMessage(
        content=(
            "你是一名严谨的技术助手。"
            "不知道时应明确说明，不得虚构 API 或参数。"
        )
    ),
    HumanMessage(content="LangChain 的 invoke 和 stream 有什么区别？"),
]

full_response = None

for chunk in model.stream(
    messages,
    config={
        "tags": ["technical-qa"],
        "metadata": {"channel": "blog-demo"},
    },
):
    full_response = chunk if full_response is None else full_response + chunk
    print(chunk.content, end="", flush=True)

if full_response is not None:
    print("\n\nToken 用量：", full_response.usage_metadata)

这个示例仍然只是模型调用层。继续构建应用时，可以在模型前加入 ChatPromptTemplate，在模型后加入 StrOutputParser 或结构化输出，再通过 LCEL 组合成完整工作流。

十六、常见误区

1. 认为 temperature=0 就绝对确定

低温度通常减少随机性，但服务端实现、模型版本、并行计算和系统更新都可能影响结果。需要严格复现时，还应固定模型快照、参数、输入和依赖版本，并接受提供商可能无法保证位级一致。

2. 认为 ainvoke 会让单次请求更快

异步调用的主要价值是等待期间不阻塞其他任务。单个请求的模型推理时间通常不会因此缩短。

3. 不限制异步并发

一次创建大量任务容易触发速率限制、连接耗尽和费用突增。并发必须受控，并为部分失败设计重试与汇总策略。

4. 只打印 content，忽略结束原因

回答可能因为输出上限、内容策略或工具调用而结束。关键流程应检查消息元数据，而不是默认 .content 就是完整最终答案。

5. 把 OpenAI 兼容端点当作完全等价

兼容端点通常只实现协议的一部分。流式输出、工具调用、图像、结构化输出和用量字段都可能存在差异，应通过集成测试确认。

6. 在代码或课件中暴露密钥

API Key 一旦出现在代码、截图、日志或版本历史中，就应视为已经泄露并立即轮换。示例只能使用占位符，不能展示真实密钥。

结语

LangChain Model I/O 的核心价值不是让一次请求少写几行代码，而是为模型调用建立稳定的工程边界：消息对象规范输入，AIMessage 承载内容与元数据，Runnable 接口统一同步、异步、流式和批量调用，模型集成则屏蔽一部分提供商差异。

掌握模型调用之后，还应继续理解提示词模板与结构化输出。三者结合起来，才能形成完整的数据流：应用以明确结构准备输入，模型执行推理，输出经过解析和校验后再进入业务系统。

参考资料

• LangChain Models
• LangChain Messages
• LangChain Python API Reference
• Ollama 官方网站
• OpenAI Tokenizer