纲要

• LangChain 大模型组件概述
• LLMs 与 ChatModels 的核心区别
• 官方合作包与社区包的依赖组织
• 大模型标准参数配置
• 标准事件驱动大模型交互
  • invoke：同步调用
  • stream：流式输出
  • batch：批量处理
  • astream_events：异步事件流
  • with_structured_output：结构化输出
• 消息格式：OpenAI 格式与 LangChain 格式
• 上下文窗口与速率限制的基础认知
• 完整可运行代码示例

为什么需要统一的大模型组件

当前大模型生态中，不同的厂商提供各自独立的 API 与调用方式。如果直接使用原生 SDK 进行 AI 应用开发，在需要切换或组合多个模型时会面临极高的集成成本。LangChain 提供的大模型组件正是为了抹平这些差异：无论是 OpenAI、DeepSeek 还是 Anthropic 的模型，都可以通过统一接口进行调用，大幅降低开发复杂度。

LLMs 与 ChatModels 的区别

LangChain 对大模型的封装分为两类：

类型	输入/输出	功能特性	当前地位
LLMs	纯字符串	仅文本补全，无工具调用、结构化输出、多模态	旧范式，逐渐弃用
ChatModels	消息列表（支持 OpenAI 与 LangChain 两种消息格式）	支持工具调用、结构化输出、多模态、流式等	主流推荐使用

• ChatModels 既兼容 OpenAI 的 system / user / assistant 消息结构，也支持 LangChain 自定义的 SystemMessage、HumanMessage、AIMessage、ToolMessage 等格式。
• 在 LangChain 中，带有 Chat 前缀的类（如 ChatOpenAI）都属于 ChatModels，而早期以纯文本为主的模型封装通常以 LLM 结尾（如 OpenAI）。目前实际开发中几乎全部采用 ChatModels。

包的组织：官方合作包与社区包

从 LangChain 0.3 版本开始，依赖包进行了拆分：

• 官方合作包：以 langchain- 为前缀，如 langchain-openai、langchain-anthropic、langchain-google-genai、langchain-aws。由 LangChain 官方维护，接口升级及时，稳定性高。
• 社区包：集中在 langchain-community 中，由开源社区贡献，维护时效和稳定性相对较低。

推荐优先使用官方合作包。安装方式示例：

pip install langchain-openai langchain-anthropic langchain-deepseek

标准参数一览

实例化大模型组件时可配置的核心参数（所有官方合作包强制统一）：

参数	说明	示例
model	模型名称	"gpt-4"
temperature	随机度（0~1），越小越遵循指令，越大越有创造性	0.0（严格）或 0.7（创意）
timeout	单次请求超时秒数	30
max_tokens	最大输出 token 数	200
stop	停止符，遇到该字符时立即停止输出	"\n" 或 "我"
max_retries	API 调用失败后的最大重试次数	3
api_key	对应厂商的 API 密钥	可从环境变量读取
base_url	代理地址（如有需要）	"https://api.openai.com/v1"
rate_limit	请求速率限制（QPS）	按厂商文档设置

注意：标准参数只对厂商 API 开放了相应能力的模型有效。社区包不强制使用统一参数名。

标准事件驱动

大模型组件提供了一系列标准方法（事件）来驱动交互，下面的时序图展示了几种核心事件的生命周期：

invoke：同步调用

最基础的调用方式，传入提示词，等待模型返回完整结果。

stream：流式输出

逐 token 返回，前端可以呈现打字机效果，大幅提升用户体验。使用 for 循环消费 chunk.content。

batch：批量处理

同时向模型发送多个请求，全部完成后统一返回结果列表。适合批量评测、离线生成等场景。

astream_events：异步事件流

基于事件的流式输出，允许开发者在模型开始响应、流式输出中、输出结束时捕获不同事件并执行自定义逻辑。需使用 async for，并指定 version="v2"。

with_structured_output：结构化输出

让模型直接返回结构化数据（如 JSON），而非纯文本。通常结合 Pydantic 定义输出模式，下游代码可直接按字段读取，避免正则解析。

消息字段说明

大模型返回的 AIMessage 包含以下关键字段：

• content：模型响应的文本内容
• tool_calls：工具调用信息（标准化字段）
• invalid_tool_calls：无效的工具调用
• usage_metadata：输入 / 输出 token 统计
• id：消息唯一标识
• response_metadata：响应头、token 计数等原始数据

不同模型的原生返回字段并不相同，LangChain 会对 usage_metadata 等做标准化处理，但部分字段仍需注意厂商差异。

完整可运行代码

以下代码展示了使用 langchain-openai 调用 GPT-4，并演示标准参数配置以及五个核心事件。运行前请设置环境变量 OPENAI_API_KEY，并安装依赖：

pip install langchain-openai pydantic

import os
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI

# 1. 配置模型，演示所有标准参数
llm = ChatOpenAI(
    model="gpt-4",
    temperature=0.4,          # 随机度
    timeout=30,               # 超时秒数
    max_tokens=200,           # 最大输出 token 数
    # stop=["我"],            # 可选停止符
    max_retries=3,            # 最大重试次数
    api_key=os.getenv("OPENAI_API_KEY"),   # 从环境变量读取
    # base_url="https://your-proxy.com/v1", # 如需代理
)

prompt = "用一句话介绍一下你自己。"

# 2. invoke：同步调用
print("=== invoke ===")
result = llm.invoke(prompt)
print(result.content)
print()

# 3. stream：流式输出
print("=== stream ===")
for chunk in llm.stream(prompt):
    print(chunk.content, end="", flush=True)
print("\n")

# 4. batch：批量处理
print("=== batch ===")
questions = [
    "LangChain 是什么？",
    "LangChain 有哪些竞品？",
]
results = llm.batch(questions)
for q, r in zip(questions, results):
    print(f"Q: {q}\nA: {r.content}\n")
print()

# 5. astream_events：异步事件流
import asyncio

async def stream_events():
    print("=== astream_events ===")
    events = []
    async for event in llm.astream_events(prompt, version="v2"):
        events.append(event)
        event_name = event["event"]
        if event_name == "on_chat_model_start":
            print(f"[事件] 模型开始响应")
        elif event_name == "on_chat_model_stream":
            chunk = event["data"]["chunk"]
            if chunk.content:
                print(chunk.content, end="", flush=True)
        elif event_name == "on_chat_model_end":
            print(f"\n[事件] 模型响应结束")
    # 最终输出包含 usage_metadata
    final_output = events[-1]["data"]["output"]
    print(f"Token 用量: {final_output.usage_metadata}")
    print()

asyncio.run(stream_events())

# 6. with_structured_output：结构化输出
print("=== structured output ===")
class Joke(BaseModel):
    """让模型输出一个笑话"""
    setup: str = Field(description="笑话的开场/包袱铺垫")
    punchline: str = Field(description="笑话的包袱/笑点")
    rating: int | None = Field(default=None, description="好笑程度 1-10 分")

structured_llm = llm.with_structured_output(Joke)
joke_result = structured_llm.invoke("给我讲一个程序员的笑话")
print(f"setup: {joke_result.setup}")
print(f"punchline: {joke_result.punchline}")
print(f"rating: {joke_result.rating}")

运行说明：

• 代码依赖 langchain-openai 和 pydantic，请先通过 pip 安装。
• 需要有效 OpenAI API Key 并设置为环境变量 OPENAI_API_KEY。
• stop 参数可根据需要取消注释，观察停止符效果。
• 异步部分需 Python 3.8+ 并安装 asyncio（标准库自带）。

总结

本文介绍了 LangChain 大模型组件的核心知识，旨在帮助开发者统一调用不同厂商的大模型。主要涵盖以下内容：

• 组件分类：区分了旧式 LLMs（纯文本补全）与推荐的 ChatModels（支持消息、工具调用、结构化输出、多模态）。
• 包管理：说明 LangChain 0.3 起拆分为官方合作包（如 langchain-openai）和社区包，并给出安装示例。
• 标准参数：罗列 model、temperature、timeout、max_tokens、stop、max_retries 等统一配置项，并提醒注意厂商差异。
• 核心事件：
  • invoke – 同步返回完整结果
  • stream – 逐 token 流式输出
  • batch – 批量并发请求
  • astream_events – 异步事件流，可监听开始、流式、结束状态
  • with_structured_output – 结合 Pydantic 直接输出结构化数据（如 JSON），避免正则解析
• 消息格式：AIMessage 包含 content、tool_calls、usage_metadata 等重要字段，LangChain 做了部分标准化。
• 代码：提供了配置 ChatOpenAI 并演示所有核心事件的可运行 Python 脚本，涵盖异步调用和结构化笑话生成示例，代码已确认可直接执行。