一、Model I/O 介绍

1.1 什么是Model I/O

Model I/O——LangChain 中与大语言模型交互的核心流程。

Model I/O 回答的是一个最基本的问题：“怎么把问题喂给模型，并拿到有用的结果？”

1.2 Model I/O 的三个环节

环节	做什么
Prompts（提示词模板）	把用户输入和系统指令格式化成模型能理解的消息，解决怎么问的问题
Model（模型调用）	统一接口调用不同平台（供应商）的模型，解决问谁的问题
Output Parsers（输出解析）	把模型返回的文本转换为JSON、Pydantic对象等结构化数据，结构化输出，解决怎么回答的问题

1.3 LangChain 中的两类模型区分

1. Chat Models：对话型模型，我们调用的LLM（大语言模型）就是它，是目前的主流模型，（例如 GPT-4o、Claude、DeepSeek）。
2. Embeddings：向量型模型，这种模型将文本转换为数字向量，在根据RAG是会使用到它，（例如 bge-m3）。

二、LLM模型实例化

2.1 为什么不使用厂商原生SDK？

使用SDK调用的缺陷：

1. 每家厂商 API、参数、返回格式不统一，换模型就要全改代码；

2. 无统一工具调用、RAG、记忆、Agent、流式封装，业务逻辑重复写；

3. 无标准化状态、回调、重试、日志、合并消息逻辑，每套模型单独维护一套。

LangChain 调用优势：

1. 统一接口，切换 OpenAI / 通义 / 智谱 / BGE 仅改一行实例；
2. 内置工具、提示词、向量库、LangGraph 流程整套封装，不用重复造轮子；
3. 统一消息结构、流式输出、错误重试、上下文合并，适配 RAG/Agent 全流程；
4. 兼容 LangGraph、MCP 生态，做复杂智能体天然适配。

总结：

​ 原生 SDK 只解决 “发请求拿结果”；LangChain 是上层标准抽象，统一多模型、配套 RAG/Agent 全业务能力，降低维护与迁移成本。

2.2 LLM在LangChain中的两种实例化方式

1. 使用LangChain集成的模型提供商类

前置环境配置

# 使用uv按需安装模型提供商集成（选你要用的）
uv add langchain-openai # OpenAI（GPT-4 等）
uv add langchain-anthropic # Anthropic（Claude 系列）
uv add langchain-google-genai # Google Gemini
uv add langchain-deepseek # DeepSeek
uv add langchain-ollama # Ollama 本地模型

# .env 环境变量加载
uv add python-dotenv

然后在将kpi-key、base-url写在.env中通过python-dotenv导入

示例代码

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

# load_dotenv() 默认行为：系统已有环境变量 > .env 文件
# override=True：.env 里的变量强制覆盖系统环境变量
# 没有环境变量可以不用写
load_dotenv(override=True)

# 导入 api-key 和 base-url
API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash', # 模型名字
    api_key=API_KEY, # 模型 api-key
    base_url=BASE_URL # 模型 base-url
)
# 调用模型
response = llm.invoke('你是谁？')
# llm.invoke() 返回的是 AIMessage 对象，不是字符串。要拿文本内容需要 .content
print(response.content)

ChatOpenAI 是在 LangChain 中最常用的类，日常开发 90% 的场景都用它。因为大部分厂家都支持OpenAI的这种实例化方式。

核心参数详解

初始化 ChatOpenAI 时可以传入多个参数来控制模型行为：

llm = ChatOpenAI(
    model="gpt-4o-mini", # 模型名称
    api_key="API_KEY", # 模型api-key
    base_url="https://api.openair.com/", # 模型base_url
    temperature=0.7, # 随机性，0=确定性，1=有创意（默认因模型而异）
    max_tokens=1000, # 最大输出长度
    timeout=60, # 超时时间（秒）
    max_retries=2, # 失败重试次数
)

temperature怎么选择

场景	推荐 temperature	原因
代码生成、数据提取、翻译	0 ~ 0.3	需要准确、稳定的输出
问答、摘要、分析	0.3 ~ 0.7	兼顾准确性和流畅性
创意写作、头脑风暴、起名	0.7 ~ 1.0	需要多样性和创造力

Token是什么

最容易误解的一点就是：它不是“字数”，也不是“单词数“。

​ 大模型真正处理的最小单位，是 token。可以把它理解成：模型内部用来读写文本的“最小片段”。这个片段有时候是一个字，有时候是半个词，有时候是一个完整单词，甚至可能只是一个标点。

​ 所以，同样一句话，人眼看起来长度差不多，token 数却可能差很多。

​ 注意：同一段文本，在不同模型、不同分词器下，token 数并不一定相同。

为什么？

​ 因为每家模型厂商背后的分词器（tokenizer）不一样，分词器本质上就是：把一段文本切成 token 的规则和词表。不同分词器的词表不同、切分策略不同，所以最后统计出来的 token 数也会不同。

​ 模型提供商通常按 token 数量计费， max_tokens 参数限制的是输出的最大 token 数。如果你发现模型的回答被截断了，通常是 max_tokens 设太小了。

补充理解：

• 输入token：你发给模型的提示词、上下文、历史对话
• 输出 token：模型生成的回答
• 总消耗token = 输入 token + 输出 token

所以，一次调用是否贵，不只取决于回答长不长，也取决于你喂给模型的上下文有多长。

2. 统一工厂 init_chat_model（推荐多模型切换项目）

​ 需要在运行时动态切换不同提供商的模型时， init_chat_model 比 ChatOpenAI 更方便——不需要 import 不同的类。

代码示例

# init_chat_model 工厂模式
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model

# override=True 让.env优先级高于系统环境变量
load_dotenv(override=True)

# 读取环境变量
API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 统一工厂创建LLM实例
llm = init_chat_model(
    # 模型名
    model="deepseek-v4-flash",
    # 模型供应商
    model_provider="openai",  # deepseek兼容OpenAI协议，可以填openai
    api_key=API_KEY,
    base_url=BASE_URL,
    temperature=0.5
)

# 同步调用
response = llm.invoke("你是谁？")
print(response.content)

init_chat_model 工厂使用注意点

1. 必须正确填写 model_provider
   支持：openai / zhipu / qwen / anthropic / ollama 等，填错会抛找不到厂商异常。
2. 环境变量规范
   工厂自动读取对应厂商KEY：OPENAI_API_KEY、ZHIPUAI_API_KEY，不要自定义变量名。统一参数名 base_url，所有厂商通用，不用记各厂商不同参数名。也可以自己手动传参定义。
3. 厂商独有高级参数无法通过工厂传入
   如智谱特有工具扩展、OpenAI独有parallel_tool_call，这种场景必须换回原生ChatXXX类。
4. 版本依赖
   通过工厂模式导入模型，虽然不需要再导入提供商类，但是还是要安装对应的包，负责工厂无法加载。例如：智谱/通义 需要额外安装对应包：pip install langchain-zhipuai。

3. 两者如何选择

1. 固定一家厂商、需要底层私有参数 → 原生ChatOpenAI或其他原生类；
2. 多模型动态切换、配置驱动、统一管理模型 → init_chat_model 工厂；
3. 业务调用（invoke/stream/ainvoke）两类实例完全通用，接口无区别。

三、模型调用方式详解

调用大模型就像打电话：你得先知道 “说什么”（消息类型），再知道 “怎么说”（传入方式），最后知道 “怎么拨号”（调用方式）。

调用大模型 = 打电话

1. 消息类型 = 区分谁说话（系统/用户/AI/工具）
2. 传入方式 = 消息打包格式（字符串/消息对象/元组）
3. 执行调用 = 拨号模式（invoke一次性/stream流式/batch批量）

3.1 消息类型

1. 核心概念

LangChain 所有对话、记忆、Agent 底层都依赖 BaseMessage 四类标准消息，模型接口强制区分角色，缺一不可。

消息类型	类名	核心作用	业务场景
SystemMessage	系统消息	最高优先级，设定AI人设、输出约束、规则	固定助手身份、限制回答格式
HumanMessage	用户消息	人类提问、输入内容	用户对话、检索指令
AIMessage	AI消息	模型历史回复，承载上下文、需要调用的工具相关信息	多轮对话记忆存储
ToolMessage	工具消息	外部工具返回数据（检索/接口计算）	RAG、Agent工具调用

记忆口诀：系统定规则，用户提问题，AI存历史，工具返数据

2. 可运行基础代码

import os
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

# 单条用户消息列表传入
response = llm.invoke([HumanMessage(content="你好")])
# 返回 AIMessage 对象，content 取文本内容
print("AI回复：", response.content)
# 拓展：打印底层对象结构，看懂返回元数据
print("返回对象类型：", type(response))
print("模型调用元数据：", response.response_metadata)

3. 多轮对话完整示例

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

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

# 对话列表严格按聊天时序从上到下排列
conversation = [
    SystemMessage(content="你是耐心的LangChain讲师，回答简洁"),
    HumanMessage(content="你好，我叫大迫杰"),
    AIMessage(content="你好大迫杰，有什么LangChain问题我可以解答？"),
    HumanMessage(content="我叫什么名字？"),
]

response = llm.invoke(conversation)
print(response.content)

4. 为什么必须拆分4类消息（底层原理）

1. 大模型原生OpenAI接口强制区分 role，LangChain只是标准化封装；
2. 记忆组件 RunnableWithMessageHistory 仅识别这四类消息对象，存读对话全靠类型判断；
3. Agent工具调用依赖 ToolMessage 传递检索结果，不分类型会直接解析失败；
4. 系统消息独立拆分，可动态切换AI人设，不用修改用户输入文本。

3.2 传入方式

ChatOpenAI.invoke() 入参仅支持三类：

1. 纯字符串 str

2. 标准消息对象列表 list[BaseMessage]

3. 角色元组列表 list[tuple]
   原生字典不属于允许类型，无法自动解析。

   需求场景	推荐方式	底层原理	优缺点
   临时单轮测试、无角色无上下文	纯字符串	内部自动封装 HumanMessage	优点：极简；缺点：不能加system、无多轮历史
   生产业务：RAG/对话机器人/Agent	消息对象列表 [SystemMessage, HumanMessage...]	LangChain标准Runnable协议，全组件兼容	优点：稳定、支持记忆、工具调用；缺点：代码略长
   快速调试、简易接口转换	角色元组 ("system", "xxx")	内部自动转换消息对象	优点：简写；缺点：不适合持久化存储

方式1：直接传入字符串（仅测试用）

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

# 内部自动转为 [HumanMessage(content="xxx")]
response = llm.invoke('你是谁？')

print(response.content)

方式2：消息对象列表（生产首选）

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

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

res = llm.invoke([
    SystemMessage(content="Python编程讲师，回答精简"),
    HumanMessage(content="什么是装饰器？")
])
print(res.content)

方式3：角色元组（临时调试）

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

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

tuple_msg = [
    ("system", "Python编程助手"),
    ("user", "什么是装饰器？")
]
res = llm.invoke(tuple_msg)
print(res.content)

补充：字典正确处理方案（外部JSON/数据库对话转标准消息）

from langchain_core.messages import SystemMessage,HumanMessage

# 模拟前端/数据库读取的原始字典对话
raw_dict_messages = [
    {"role": "system", "content": "你是翻译助手"},
    {"role": "user", "content": "解释机器翻译"}
]

# 角色映射：字典role → LangChain标准消息类,可以补充上ai、tool
role_mapping = {
    "system": SystemMessage,
    "user": HumanMessage
}
# 批量转换为合法消息对象列表
message_list = [role_mapping[msg["role"]](content=msg["content"]) for msg in raw_dict_messages]

print(message_list)
# message_list
# [SystemMessage(content='你是翻译助手', additional_kwargs={}, response_metadata={}), HumanMessage(content='解释机器翻译', additional_kwargs={}, response_metadata={})]

3.3 调用方式

LangChain 所有模型都遵守统一的 Runnable 协议，固定提供 4 类成对接口：
同步：invoke / stream / batch
异步：ainvoke / astream / abatch
外加高级事件流 astream_events，全部接口一套逻辑，只是适配不同业务场景。

类比理解：给大模型发消息像网购下单

1. invoke：只买一件，付完钱等完整包裹送达
2. stream：商品分多个快递盒逐件发货，到一件拆一件
3. batch：一次性下单多件商品，平台统一批量发货
4. ainvoke/abatch：同时下多单，不用等上一单完成再下单

1. 同步调用 invoke() （最常见）

核心概念

同步 = 阻塞调用：代码执行到 llm.invoke() 会卡住，必须等大模型完整返回所有文字，才会执行下一行代码。

底层逻辑

内部流程：组装消息 → 发起HTTP请求 → 阻塞等待网络返回完整响应 → 封装成 AIMessage 对象返回。

标准代码

from langchain_openai import ChatOpenAI

# 实例化聊天模型
llm = ChatOpenAI()
# 发起同步调用，支持字符串/消息列表/元组三种入参
response = llm.invoke("什么是LangChain？")
# response 是 AIMessage 对象，.content 提取纯文本
print(response.content)

适用场景 & 优缺点

优点：代码最简单、无异步语法门槛，调试方便
缺点：串行阻塞，大量请求排队等待，耗时叠加
适用：本地脚本、单次问答、小流量后台同步接口。

2. 异步调用 ainvoke()（高并发场景）

前置通俗讲解：同步 vs 异步

模型调用绝大多数耗时都在网络等待，CPU全程空闲。

• 同步 invoke：串行排队，一个请求没返回，下一个永远发不出去
  5个请求，每个2秒，总耗时≈10秒
• 异步 ainvoke：释放CPU，等待网络时可以同时发起其他请求
  5个请求并行，总耗时≈最慢的那一个请求（2秒）

奶茶店类比：
同步：你买完一杯，等做好，再买下一杯；
异步：一次性扫码点5杯，后厨同时制作，不用挨个等。

关键语法规则

1. 异步函数定义必须加 async def
2. 调用异步方法 ainvoke 必须加 await
3. 普通py文件用 asyncio.run() 启动；Jupyter 直接 await

基础可运行代码拆解：

import asyncio
from langchain_openai import ChatOpenAI

llm = ChatOpenAI()

# 定义异步函数
async def call_llm_async():
    # await 等待异步请求完成，不会阻塞整个程序
    response = await llm.ainvoke("什么是LangChain？")
    print(response.content)

# 普通 .py 文件固定启动写法
if __name__ == "__main__":
    asyncio.run(call_llm_async())

并行加速核心：asyncio.gather

误区纠正（重点）

只单独循环 await ainvoke 不会提速，依旧串行：

# 错误写法：串行，无并发加速
async def wrong_demo():
    r1 = await llm.ainvoke("问题1")
    r2 = await llm.ainvoke("问题2")

原因：必须等第一个请求完全结束，才会创建第二个请求。

正确并行写法

asyncio.gather 会收集所有协程，一次性全部派发，同时等待所有请求返回：

async def right_demo():
    # 仅创建任务，不等待
    task_list = [
        llm.ainvoke("问题1"),
        llm.ainvoke("问题2"),
        llm.ainvoke("问题3"),
    ]
    # *tasks 解包列表，一次性并发执行全部任务
    results = await asyncio.gather(*task_list)

同步/异步完整对比代码：

import time
import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)
prompts = ["简洁介绍北京", "简洁介绍上海", "简洁介绍广州", "简洁介绍深圳", "简洁介绍杭州"]

# 同步串行测试
def test_sync_invoke():
    print("=== 同步串行 ===")
    start = time.time()
    for p in prompts:
        res = llm.invoke(p)
        print(f"【问题】{p}\n【回答】{res.content}\n")
    print(f"耗时：{time.time() - start:.2f}s\n")

# 异步并行测试（修改重点：接收gather返回的results列表）
async def test_async_ainvoke():
    print("=== 异步并行 ===")
    start = time.time()
    tasks = [llm.ainvoke(p) for p in prompts]
    # 关键：用变量接收 gather 的返回值
    results = await asyncio.gather(*tasks)

    # results 是 AIMessage 对象列表，和 prompts 一一对应
    for question, msg in zip(prompts, results):
        print(f"【问题】{question}")
        print(f"【回答】{msg.content}\n")

    print(f"耗时：{time.time() - start:.2f}s\n")

async def main():
    # test_sync_invoke()
    await test_async_ainvoke()

if __name__ == "__main__":
    asyncio.run(main())

运行效果：同步总耗时约29秒，异步仅7秒左右，请求越多差距越大。

生产避坑

一次性 gather 上百个请求会触发 OpenAI 429 限流，大批量任务需要用 asyncio.Semaphore 限制最大并发数量。

适用场景

FastAPI 异步Web接口、批量数据处理、多模型同时对比、高并发SaaS系统。

3. 流式调用 stream() / astream()（前端打字机效果）

核心原理

普通 invoke 要等模型生成完整文本再返回；
stream 让模型生成一个字（token）就立刻返回一段分片 AIMessageChunk，前端实时逐字打印，用户不用长时间空白等待。

标准同步流式代码

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

def stream_demo():
    print("AI实时输出：")
    full_msg = None
    # stream 返回迭代器，循环读取每一小段文字
    for chunk in llm.stream("写一首春日短诗"):
        # 官方规范：用 += 拼接分片，兼容性最强，不会报类型错误
        if full_msg is None:
            full_msg = chunk
        else:
            full_msg += chunk
        # flush=True 关闭缓冲区，文字立刻打印到控制台
        print(chunk.content, end="", flush=True)
    # 循环结束后，full_msg 是完整的消息对象
    print("\n完整文本：\n", full_msg.content)

stream_demo()

高级事件流 astream_events（链路监控）

stream 只能拿到文字分片；astream_events 可以捕获全生命周期事件：模型开始、分片输出、模型结束，适合日志、链路追踪。
强制要求：必须传 version="v2"，新版LangChain硬性约束。

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

async def stream_event_demo():
    async for event in llm.astream_events("介绍LCEL", version="v2"):
        event_name = event["event"]
        # 过滤无关事件，只保留关键三类，避免日志爆炸
        if event_name == "on_chat_model_start":
            print(f"\n【模型开始生成】")
        elif event_name == "on_chat_model_stream":
            print(event["data"]["chunk"].content, end="", flush=True)
        elif event_name == "on_chat_model_end":
            print("\n【生成完毕】")

if __name__ == "__main__":
    asyncio.run(stream_event_demo())

适用场景

Web聊天机器人、长文本生成、SSE实时推送、Agent工具调用过程可视化。

4. 批次调用 batch() / abatch()（批量处理任务）

概念

一次性传入多条独立问题，框架内部自动调度并发执行，批量返回结果列表，不用自己写循环。

同步 batch 基础示例

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

def batch_demo():
    questions = ["简介回答什么是Python？", "简介回答什么是JS？", "简介回答什么是Go？"]
    # 批量执行，返回和输入等长的响应列表
    responses = llm.batch(questions)
    for q, res in zip(questions, responses):
        print(f"问题：{q}\n回答：{res.content}\n")

batch_demo()

异步 abatch

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()

API_KEY = os.getenv('DEEPSEEK_API_KEY')
BASE_URL = os.getenv('DEEPSEEK_BASE_URL')

# 创建模型实例
llm = ChatOpenAI(
    model='deepseek-v4-flash',
    api_key=API_KEY,
    base_url=BASE_URL
)

async def abatch_demo():
    questions = ["简介回答LangChain是什么？", "简介回答LCEL作用？", "简介回答RAG流程？"]
    responses = await llm.abatch(questions)
    for q, res in zip(questions, responses):
        print(f"{q}\n{res.content}\n")

if __name__ == "__main__":
    asyncio.run(abatch_demo())

生产避坑

batch/abatch 无内置并发限制，大批量文档清洗、摘要时，必须手动限制并发，防止接口限流报错429。

适用场景

批量翻译、批量文档摘要、数据集批量标注、离线数据清洗。

3.4调用配置与高级特性教学

1. config 运行时配置 + 回调系统

核心区分三个概念（重点，极易混淆）

1. 输入input/messages：发给大模型的业务提问内容（包裹里的货物）
2. config 字典：元数据，仅用于监控、日志、分类，不会传给大模型（快递面单标签）

• • tags：自定义分类标签，LangSmith监控平台筛选用
  • metadata：自定义业务字段（user_id、session_id）

3. callbacks 回调处理器：监听模型全生命周期事件（独立监控摄像头，不能放进config）

标准规范写法（分开传参）

from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler

# 自定义回调：拦截模型启动事件，打印监控信息
class MyDebugCallback(BaseCallbackHandler):
    def on_chat_model_start(self, serialized, prompts, **kwargs):
        print("\n==== 请求监控 ====")
        print(f"模型：{serialized['kwargs']['model']}")
        print(f"标签：{kwargs.get('tags')}")
        print(f"用户ID：{kwargs.get('metadata')['user_id']}")
        print("==================\n")

llm = ChatOpenAI(model="gpt-4o-mini")

resp = llm.invoke(
    input="讲一个编程短笑话",
    # config 只存放tags、metadata
    config={
        "tags": ["demo", "debug"],
        "metadata": {"user_id": "user_001", "session_id": "s12345"}
    },
    # 回调独立参数传入，标准写法
    callbacks=[MyDebugCallback()]
)
print(resp.content)

2. 运行时动态切换模型 init_chat_model

原理讲解

init_chat_model 是通用模型初始化工具，创建一个可配置的通用模型实例，不用重新new对象，单次调用通过 config["configurable"] 切换不同大模型。

from langchain_init import init_chat_model

# 全局基础模型，统一温度等基础参数
base_llm = init_chat_model(temperature=0)

# 调用1：切换 gpt-4o-mini
res1 = base_llm.invoke(
    "你好",
    config={"configurable": {"model": "gpt-4o-mini"}}
)
print("gpt4o-mini：", res1.content)

# 调用2：运行时切换 Claude
res2 = base_llm.invoke(
    "你好",
    config={"configurable": {"model": "claude-3-sonnet-4-6"}}
)
print("claude：", res2.content)

业务落地场景

1. 前端页面让用户自主选择模型；
2. 模型A/B测试，对比不同模型输出质量；
3. 多租户SaaS平台，给不同客户分配不同厂商大模型。

四、扩展

4.1 多模态输入（VLM 视觉语言模型）

1. 基础概念

普通LLM：只能处理纯文字；
VLM（视觉语言大模型）：同时接收文字 + 图片，既能看懂图片画面，又能用文字描述画面内容。
代表模型：GPT-4o、Claude 全系、DeepSeek-VL、Qwen-VL、通义千问VL。

2. 核心传输规则

1. 远程公网图片：直接填 https://xxx.jpg 字符串，不用编码，最简单；
2. 本地电脑图片：不能直接传文件路径，API服务访问不到你本地磁盘，必须转 Base64 文本格式；
3. 消息固定结构：HumanMessage 的 content 传列表，分别放文本块、图片块。

3. Base64 通俗原理

图片是二进制字节，HTTP/JSON 只能传输文本字符，二进制会乱码、报错。
Base64 把图片二进制转成纯 ASCII 字符串，包装成 data:image/jpeg;base64,xxxx 格式（Data URL），模型API能识别这串文本代表一张图片。

4. 完整可运行代码

import base64
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

load_dotenv()

# 初始化国产多模态模型 DeepSeek-VL
llm = ChatOpenAI(
    model="deepseek-vl",
    api_key=os.getenv("DEEPSEEK_VL_API_KEY"),
    base_url=os.getenv("DEEPSEEK_VL_BASE_URL")
)

# 工具函数：本地图片转base64 DataURL，封装异常
def img_to_base64(img_path: str) -> str:
    try:
        with open(img_path, "rb") as f:
            img_bin = f.read()
        # 编码bytes → 字符串
        b64_str = base64.b64encode(img_bin).decode("utf-8")
        return f"data:image/jpeg;base64,{b64_str}"
    except FileNotFoundError:
        print(f"错误：文件 {img_path} 不存在")
        raise
    except Exception as e:
        print(f"图片编码失败：{e}")
        raise

# 方式1：读取本地图片
local_img_url = img_to_base64("test.jpg")
msg_local = HumanMessage(content=[
    {"type": "text", "text": "详细描述这张图片里所有内容"},
    {"type": "image_url", "image_url": {"url": local_img_url}}
])

# 方式2：远程网络图片（无需编码，线上业务优先用）
msg_remote = HumanMessage(content=[
    {"type": "text", "text": "图中是什么物体？"},
    {"type": "image_url", "image_url": {"url": "https://xxx/demo.jpg"}}
])

# 调用模型识别图片
res = llm.invoke([msg_local])
print("图片识别结果：\n", res.content)

5. 高频踩坑点

1. 图片体积过大 → API 返回报错，前端提前压缩图片；
2. 传入 ./test.jpg 文件路径，而非base64 → 服务商读取不到本地文件，识别失败；
3. 国产VLM部分不支持webp格式，统一转jpg/png。

4.2 速率限制（限流，解决429请求过多报错）

1. 为什么需要限流？

所有大模型厂商都限制每秒/每分钟最大请求数（QPS），短时间并发大量请求，接口直接返回 429 Too Many Requests，程序中断。限流组件用来控制请求发送速度。

2. 同步限流：InMemoryRateLimiter

仅适用于单进程同步脚本，内存存储令牌桶，多进程、异步、分布式项目完全失效。
参数讲解：

• requests_per_second=0.1：每秒允许0.1次请求 → 每10秒只能发1条；
• check_every_n_seconds=0.1：每0.1秒检查一次是否有可用令牌。

import time
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.rate_limiters import InMemoryRateLimiter

load_dotenv()

# 构建令牌桶限流器
rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.1,
    check_every_n_seconds=0.1
)

llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    rate_limiter=rate_limiter
)

def test_limit(n=3):
    print("任务开始", time.strftime("%X"))
    last_time = time.time()
    for i in range(n):
        t0 = time.time()
        resp = llm.invoke(f"第{i}次请求，简单回复一句话")
        t1 = time.time()
        print(f"第{i}轮 | 单次耗时{t1-t0:.2f}s | 间隔上一次{t1-last_time:.2f}s")
        last_time = t1

test_limit(3)

运行现象：两次调用强制间隔10秒，不会触发超限报错。

3. 异步项目限流

InMemoryRateLimiter 不兼容异步 ainvoke，异步统一用 asyncio.Semaphore 信号量控制最大并发数量。

import asyncio
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

# 封装带并发限制的请求函数
async def safe_request(prompt, sem):
    async with sem:  # 信号量自动控制并发
        return await llm.ainvoke(prompt)

async def async_limit_demo():
    sem = asyncio.Semaphore(2)  # 同一时间最多2个请求并发
    prompts = ["问题1", "问题2", "问题3", "问题4", "问题5"]
    task_list = [safe_request(p, sem) for p in prompts]
    results = await asyncio.gather(*task_list)
    for res in results:
        print(res.content)

asyncio.run(async_limit_demo())

4. 限流方案选型对比

方案	适用场景	短板
InMemoryRateLimiter	单机同步测试脚本	不支持异步、多进程、集群
asyncio.Semaphore	FastAPI异步接口、批量异步任务	仅单台机器生效
Redis分布式限流	多实例集群、线上生产服务	需要额外部署Redis

4.3 Token使用追踪

核心说明

usage_metadata 会记录单次调用消耗的输入Token、输出Token、总Token，仅兼容OpenAI协议接口（DeepSeek、GPT、Claude），小众模型无此字段。

方案1：临时上下文统计（简单同步场景）

适合一段代码内少量调用，自动统计上下文内所有请求总Token。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import get_usage_metadata_callback

load_dotenv()
llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

# 上下文管理器，自动收集这段代码内所有token消耗
with get_usage_metadata_callback() as cb:
    llm.invoke("什么是LCEL？")
    llm.invoke("RAG完整流程是什么？")

# 打印累计用量
print("总Token消耗：", cb.usage_metadata)
# 输出格式：{'input_tokens': xx, 'output_tokens': xx, 'total_tokens': xx}

方案2：全局自定义回调（生产推荐，同步/异步全兼容）

上下文管理器不支持异步调用，线上项目统一自定义回调，全局捕获每一次模型调用的Token。

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler

load_dotenv()

# 自定义Token统计回调类
class TokenStatCallback(BaseCallbackHandler):
    def __init__(self):
        self.all_input = 0
        self.all_output = 0

    # 模型调用结束自动触发
    def on_chat_model_end(self, response, **kwargs):
        usage = response.usage_metadata
        in_tok = usage.get("input_tokens", 0)
        out_tok = usage.get("output_tokens", 0)
        self.all_input += in_tok
        self.all_output += out_tok
        print(f"单次：输入{in_tok}，输出{out_tok}")

# 实例化回调，挂载到模型
token_cb = TokenStatCallback()
llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
    callbacks=[token_cb]
)

# 多次调用自动累计
llm.invoke("介绍LangChain")
llm.invoke("什么是Runnable")
print(f"累计总输入Token：{token_cb.all_input}")
print(f"累计总输出Token：{token_cb.all_output}")

4.4 llm.profile 探测模型能力

作用

运行时动态查看模型上限、能力开关：最大上下文长度、是否支持图片输入、是否支持工具调用、结构化输出。

关键坑

DeepSeek、通义千问等国产模型不支持profile属性，直接访问会直接报错，必须加存在判断。

兼容安全代码

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model="deepseek-v4-flash",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_BASE_URL")
)

# 安全判断，避免国产模型抛异常
if hasattr(llm, "profile") and llm.profile is not None:
    info = llm.profile
    print("模型最大上下文：", info.get("max_input_tokens"))
    print("是否支持图片输入：", info.get("image_inputs"))
    print("是否支持工具调用：", info.get("tool_calling"))
else:
    print("当前厂商未开放profile接口，无法读取模型能力配置")

业务落地场景

做多模型切换平台时，根据返回能力自动分支：不支持图片就隐藏上传按钮，不支持工具调用就禁用Agent功能。

4.5 提示词缓存 Prompt Caching（服务商侧KV缓存）

1. 核心区分两个极易混淆的缓存

① 服务商KV缓存

缓存位置：大模型厂商云端服务器
缓存内容：重复固定长系统提示词的计算KV状态
效果：相同系统词重复调用，Token计费减半/减免，延迟降低
没有本地文件，不需要自己存储任何数据
仅支持：GPT、Claude；DeepSeek、国产大模型无该功能。

② LangChain本地LLM缓存（另一独立模块）

缓存位置：内存/Redis/SQLite
缓存内容：完整「用户提问→AI回答」键值对
效果：完全不发起API请求，零Token消耗，重复问题直接本地返回
和本节完全无关，不要混淆。

2. 各平台缓存规则

平台	使用方式	成本优惠	限制
OpenAI	全自动零配置	缓存部分Token半价	连续请求系统提示完全一致、文本足够长才生效
Claude	手动引入中间件标记	缓存部分Token省90%	必须显式标记可缓存文本
DeepSeek/国产	无服务端KV缓存	无优惠	无法使用该特性

3. OpenAI自动缓存示例

连续调用携带完全相同的超长系统提示词，第二次自动触发云端缓存：

from langchain_openai import ChatOpenAI

# 超长固定系统提示
sys_prompt = """
你是专业后端讲师，回答简洁分点，必须附带可运行Python代码。
所有解释禁止长篇大段，每条知识点控制在三行以内。
后续全部提问严格遵守以上输出规范。
"""

llm = ChatOpenAI(model="gpt-4o-mini")
# 第一次：全量计费，写入云端缓存
res1 = llm.invoke([("system", sys_prompt), ("user", "什么是LCEL")])
# 第二次：系统词命中缓存，仅收取用户问题Token费用
res2 = llm.invoke([("system", sys_prompt), ("user", "什么是RAG")])

4. Claude显式缓存

from langchain_anthropic import ChatAnthropic, AnthropicPromptCachingMiddleware

llm = ChatAnthropic(
    model="claude-sonnet-4-20250514",
    middleware=[AnthropicPromptCachingMiddleware()]
)
# 长系统提示会被中间件自动标记为可缓存
resp = llm.invoke([("system", "超长固定规则..."), ("user", "你的问题")])

5. 核心避坑

1. DeepSeek等国产模型没有云端KV缓存，不要依靠它节省成本；
2. 系统提示词多一个空格、换行、文字改动，缓存直接失效；
3. 长时间间隔后，服务商云端缓存会自动清空，需要重新生成缓存；
4. 想要“相同问题不走API”，使用 langchain.cache 本地缓存，不是本节内容。