Appearance
高级特性
LangChain V1.1.0
1、多模态输入
支持图像、音频等多模态输入:
python
import base64
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model="gpt-4o") # 需要使用支持视觉语言的模型(VLM模型:视觉语言模型 LLM模型)
# 视觉模型:只能看懂图片的内容(OCR---yolox yolo-v8)
# 视觉语言模型:即能看懂图片内容 也能把看到的内容输出出来(VLM:各个平台都有各种各样的视觉模型)
# 读取图片
with open("image.jpg", "rb") as f:
image_data = f.read()
message = HumanMessage(content=[
{"type": "text", "text": "描述这张图片"},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64.b64encode(image_data).decode()}"}}
])
# type:image_url :既支持远程可以访问的图片地址 也支持本地图片(不能把本地图片路径丢给它 本地图片内容)
response = llm.invoke([message])
print(response.content)扩展:Base64 编码
概念: 将二进制数据编码为可打印文本,常用于在文本协议(如 JSON)中传输二进制数据。
编码原理: 原始二进制每个字节可以是 0-255 中的任意值,其中很多是不可打印的控制字符,无法直接放进 JSON。Base64 将每 3 个字节(24 比特)切分为 4 组(每组 6 比特),每组对应 64 个可打印字符之一(A-Z、a-z、0-9、+、/)。如果末尾不足 3 个字节,用 = 填充。
python
import base64
# 编码
with open("image.png", "rb") as f:
binary_data = f.read() # bytes: 原始二进制
base64_bytes = base64.b64encode(binary_data) # bytes: Base64 编码后仍是 bytes 类型
base64_string = base64_bytes.decode("utf-8") # str: 转成字符串(Python 类型要求)
# .decode("utf-8") 的作用:
# Python 的 b64encode 返回 bytes 类型,但 JSON/f-string 需要 str 类型
# 因为 Base64 输出全是 ASCII 字符,所以这里只是做类型转换,不涉及字符编码解读
# 解码
binary_data = base64.b64decode(base64_string)在 VLM 中的应用:
原始图片 (二进制)
│
▼ base64.b64encode()
Base64 字符串: "iVBORw0KGgoAAAANSUhEUgAA..."
│
▼ 拼接为 Data URL
"data:image/jpeg;base64,iVBORw0KGgoAAAANSUhEUgAA..."
│ │ │ │
│ │ │ └── 实际的图片数据
│ │ └── 编码方式
│ └── 媒体类型(告诉 API 这是 JPEG 图片)
└── Data URL 协议前缀2、速率限制
python
import time
from langchain_openai import ChatOpenAI
from langchain_core.rate_limiters import InMemoryRateLimiter
rate_limiter = InMemoryRateLimiter(
requests_per_second=0.1, # 10 秒才产生 1 个令牌 也即10 秒只能发1 个请求
check_every_n_seconds=0.1, # 每 0.1 秒检查一下"桶里有没有令牌可以用"
)
llm = ChatOpenAI(
model="gpt-4o-mini",
rate_limiter=rate_limiter,
)
def test_rate_limit(n=3):
print("开始时间:", time.strftime("%X"))
last = 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,"
f"距上次调用结束间隔 {t1 - last:.2f}s"
)
last = t1
test_rate_limit(3)3、Token使用追踪
python
from langchain_core.callbacks import get_usage_metadata_callback
llm = ChatOpenAI(model="gpt-4o-mini")
with get_usage_metadata_callback() as cb:
llm.invoke("你好")
llm.invoke("再见")
print(cb.usage_metadata)
# {
# 'input_tokens': 总输入token数,
# 'output_tokens': 总输出token数,
# 'total_tokens': 总token数
# }4、模型配置文件
查看模型的能力和限制:
python
llm = ChatOpenAI(model="gpt-4o")
print(llm.profile)
# {
# 'max_input_tokens': 128000,
# 'image_inputs': True,
# 'tool_calling': True,
# 'structured_output': True,
# ...
# }5、提示词缓存(Prompt Caching)
5.1 什么是提示词缓存?
当你反复用同一段很长的系统提示词(System Prompt)调用模型时,模型每次都要重新"读"这段提示词,消耗时间和 token 费用。提示词缓存让模型服务商在服务器端缓存这段已处理的提示词,后续调用直接复用,跳过重复计算。
第1次调用:
[系统提示词 3000字] + [用户问题] → 模型处理全部内容 → 💰 全价计费
第2次调用(缓存命中):
[系统提示词 3000字 ← 已缓存,跳过] + [用户问题] → 模型只处理新内容 → 💰 大幅省钱关键认知:这里的"缓存"发生在模型服务商的服务器上,不是你本地硬盘上的某个文件。你的项目目录里不会多出 cache.sqlite 之类的东西。
5.2 不同平台的缓存机制
| 平台 | 缓存方式 | 你需要做什么 | 省多少 |
|---|---|---|---|
| OpenAI | 全自动 | 什么都不用做,OpenAI 自动识别相同前缀并缓存 | 缓存命中的 token 费用减半 |
| Anthropic | 显式标记 | 用中间件告诉 API "这段可以缓存" | 缓存命中的 token 费用降低 90% |
OpenAI — 自动缓存(零配置)
python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
# 不需要任何额外配置
# OpenAI 会自动识别:如果连续多次请求的前缀(如系统提示词)相同,
# 服务器会复用已计算的 KV cache,减少延迟和费用
response = llm.invoke("你好")你在代码层面看不到"缓存存在哪"或"命中率多少",只能通过费用账单感知到缓存生效了。
Anthropic — 显式缓存(需要中间件)
python
from langchain_anthropic import ChatAnthropic, AnthropicPromptCachingMiddleware
llm = ChatAnthropic(
model="claude-sonnet-4-20250514",
# 中间件会自动把长的系统提示词标记为"可缓存"
middleware=[AnthropicPromptCachingMiddleware()]
)
# 第1次调用:全价(缓存写入)
# 第2次及之后:缓存命中,token 费用降低 90%
response = llm.invoke("你好")AnthropicPromptCachingMiddleware 的作用是:按照 Anthropic 的 API 规范,将系统提示词等长文本片段包装成带 cache_control 标记的请求,告诉 Anthropic 服务器"这段内容可以缓存"。缓存的存储和命中完全由 Anthropic 服务端负责,LangChain 只负责标记。
5.3 提示词缓存 vs 本地缓存
这是两个完全不同的东西,不要混淆:
| 维度 | 提示词缓存(Prompt Caching) | 本地缓存(LLM Cache) |
|---|---|---|
| 缓存位置 | 模型服务商的服务器 | 你本地的内存/磁盘/Redis |
| 缓存什么 | 已计算的提示词中间状态(KV cache) | 完整的"问题→回答"键值对 |
| 效果 | 减少 token 费用和延迟 | 相同问题完全不调 API,零费用 |
| 你能控制吗 | 不能 | 完全由你控制 |
| 本节讲的 | ✅ 是这个 | ❌ 不是这个 |
如果你需要"相同问题直接从本地返回、完全不打 API"的缓存,那是 LangChain 的 LLMCache 机制(如 InMemoryCache、SQLiteCache),属于另一个话题,不在本节范围内。
6、小结
本节介绍了Model I/O中的模型调用部分:
已学内容
| 主题 | 核心要点 |
|---|---|
| Model I/O概述 | Prompts → Models → Output Parsers 三部分 |
| 初始化模型 | init_chat_model 或特定包的类 |
| 消息类型 | SystemMessage, HumanMessage, AIMessage, ToolMessage |
| 调用方式 | invoke, ainvoke, stream, batch |
| 平台接入 | CloseAI代理、DeepSeek、硅基流动、init_chat_model统一切换 |
| 本地模型 | Ollama等框架 |
| 高级特性 | 多模态、速率限制、Token追踪 |
统一接口优势
python
# 同样的代码,只需更换初始化方式即可切换模型
llm = init_chat_model("gpt-4o-mini", model_provider="openai")
llm = init_chat_model("claude-sonnet-4-6", model_provider="anthropic")
llm = init_chat_model("gemini-2.5-flash", model_provider="google")
llm = ChatOllama(model="qwen2.5:7b") # 本地模型
# 调用方式完全一致
response = llm.invoke("你好")