Skip to content

高级特性

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 机制(如 InMemoryCacheSQLiteCache),属于另一个话题,不在本节范围内。

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("你好")