消息类型与调用方式

LangChain V1.1.0

1、消息类型：传什么

LangChain 用标准化的消息格式来传递不同角色的内容。理解这四种消息类型，是构建对话应用的基础。

消息类型	类名	用途	示例
系统消息	`SystemMessage`	设定AI的行为、角色和规则	"你是一个有帮助的助手"
用户消息	`HumanMessage`	用户的输入	"帮我解释一下量子计算"
AI消息	`AIMessage`	AI的回复，可用于对话历史	"量子计算是..."
工具消息	`ToolMessage`	工具执行返回的结果	工具调用的输出

记忆口诀：系统定规则，用户提问题，AI给回复，工具报结果。

在LangChain中有且仅有这四种消息类型。

基础示例：

python

from langchain_core.messages import (
    HumanMessage
)
from langchain_openai import ChatOpenAI

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

# 单个消息
response = llm.invoke([HumanMessage(content="你好")])
print(response.content)

实战示例：构建完整对话历史

python

from langchain_core.messages import (
    HumanMessage,SystemMessage,AIMessage
)
# 对话历史——模拟多轮对话
conversation = [
    SystemMessage(content="你是一个有帮助的AI助手"),
    HumanMessage(content="你好，我叫hzk"),
    AIMessage(content="你好！hzk,有什么我可以帮助你的吗？"),
    HumanMessage(content="我叫什么名字？"),
]

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

为什么需要区分消息类型？
系统消息：设定AI的"人设"和行为规则
用户消息：真正的问题或指令
AI消息：保留历史上下文，让AI"记得"之前说过什么
工具消息：当AI调用外部工具时，工具返回的结果

2、传入方式：怎么传

知道"传什么"之后，下一个问题是："怎么传给模型？"常用三种传入方式，各有适用场景。

快速决策表

你的需求	推荐方式	代码示例
简单问答，不需要上下文	直接传字符串	`llm.invoke("你好")`
需要角色设定或对话历史	传消息列表	`llm.invoke([SystemMessage(...), HumanMessage(...)])`
动态构建消息，或从其他格式转换	用元组/字典	`llm.invoke([("system", "..."), ("user", "...")])`

一句话记忆：

能用字符串就用字符串（最简单）
需要对话/角色时用消息列表（最常见）
动态构建时(动态传入变量)用元组/字典（最灵活）

方式一：直接传入字符串（最简单）

python

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
response = llm.invoke("你好，介绍一下LangChain")
print(response.content)

适用场景：

单轮问答，不需要上下文
不需要设定 AI 的角色或行为规则
快速测试或调试

优势：代码最简洁，一行搞定

局限：无法保留对话历史，无法设定系统提示词

方式二：传入消息列表（最常用）

python

from langchain_core.messages import HumanMessage, SystemMessage, AIMessage

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

response = llm.invoke([
    SystemMessage(content="你是一个专业的Python编程助手"),
    HumanMessage(content="什么是装饰器？")
])
print(response.content)

适用场景：

需要设定 AI 角色（比如"你是一个翻译助手"）
需要保留对话历史
需要区分系统指令/用户输入/AI回复

注意：这种方式是不支持动态传入变量的

实战示例：多轮对话

python

from langchain_core.messages import HumanMessage, AIMessage

# 对话历史
conversation = [
    HumanMessage(content="什么是LangChain？"),
    AIMessage(content="LangChain是一个用于开发大模型应用的框架。"),
    HumanMessage(content="它有哪些核心组件？")  # 这依赖于上一轮的上下文
]

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

方式三：使用元组或字典（最灵活）

python

# 元组方式：(角色, 内容)
tuple_messages = [
    ("system", "你是一个专业的Python编程助手"),
    ("user", "什么是装饰器？")
]

# 字典方式：{"role": 角色, "content": 内容}
dict_messages = [
    {"role": "system", "content": "你是一个专业的Python编程助手"},
    {"role": "user", "content": "什么是装饰器？"}
]

print(llm.invoke(tuple_messages))
print(llm.invoke(dict_messages))

适用场景：

从 API 返回的 JSON 数据直接转成消息列表
从配置文件或数据库读取对话模板
动态构建消息列表

实战示例：从配置读取对话模板

python

# 假设这是从配置文件读取的
prompt_template = [
    {"role": "system", "content": "你是一个{role}"},
    {"role": "user",   "content": "请解释{topic}"}
]

# 动态填充
messages = [
    {
        "role": t["role"],
        "content": t["content"].format(
            role="翻译助手",
            topic="机器翻译",
        ),
    }
    for t in prompt_template
]
print(llm.invoke(messages).content)

3、调用方式：怎么调

知道"传什么"和"怎么传"之后，最后一个问题是："怎么调用模型？"LangChain 提供了多种调用方式，适应不同场景。

3.1 同步调用 - `invoke()`（最常用）

最基础的调用方式，适合大多数场景：

python

response = llm.invoke("什么是LangChain？")
print(response.content)

适用场景：

单次调用，不需要高并发
简单问答、文本生成
快速原型开发

3.2 异步调用 - `ainvoke()`（高并发）

适用于需要同时处理多个请求的高并发场景。要理解 ainvoke() 的价值，需要先搞清楚一个前置知识：Python 的异步编程（async/await）。

前置知识：同步 vs 异步

同步（invoke）：排队买奶茶
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
你 → 点单 → 等10分钟 → 拿到 → 再点下一杯 → 等10分钟 → 拿到
     5杯奶茶总耗时：50分钟（串行等待）

异步（ainvoke）：扫码下单
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
你 → 5杯同时下单 → 5杯一起做，哪杯好了取哪杯
     5杯奶茶总耗时：≈10分钟（并行等待）

为什么模型调用特别适合异步？ 因为 llm.invoke() 的耗时几乎全花在"等网络响应"上，CPU 其实是闲着的。异步让你在等第1个响应的同时，把第2、3、4、5个请求也发出去，所有等待时间重叠，总耗时 ≈ 单次最慢的那个请求。

基础用法

python

import asyncio
from langchain_openai import ChatOpenAI

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

async def call_llm_async():
    response = await llm.ainvoke("什么是LangChain？")
    print(response.content)

# Jupyter Notebook 中直接 await（见下方说明）.py文件中不能在顶层代码中直接调用
await call_llm_async() # 方式一
asyncio.run(call_llm_async()) # 方式二

invoke() vs ainvoke() 性能对比

python

import time
import asyncio
from langchain_openai import ChatOpenAI

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

# 准备 5 个测试问题
prompts = [
    "用一句话介绍一下北京",
    "用一句话介绍一下上海",
    "用一句话介绍一下广州",
    "用一句话介绍一下深圳",
    "用一句话介绍一下杭州"
]


# ========== 测试一：同步 invoke（串行） ==========
def test_sync_invoke():
    print("=== 同步 invoke ===")
    start_time = time.time()

    for i, prompt in enumerate(prompts):
        print(f"  [同步] 正在发送第 {i + 1} 个请求...")
        llm.invoke(prompt)  # 死等，拿到结果才进入下一次循环

    print(f"总耗时: {time.time() - start_time:.2f} 秒\n")


# ========== 测试二：异步 ainvoke（并行） ==========
async def test_async_ainvoke():
    print("=== 异步 ainvoke ===")
    start_time = time.time()

    # 关键：用 asyncio.gather 同时派发所有请求
    print("  [异步] 瞬间派发 5 个请求...")
    tasks = [llm.ainvoke(prompt) for prompt in prompts] # 这不会并发执行，只是创建了协程对象列表
    results = await asyncio.gather(*tasks) #并发执行

    for r in results:
        print(f"  回答: {r.content[:20]}...")

    print(f"总耗时: {time.time() - start_time:.2f} 秒\n")


# ========== 运行对比 ==========
async def main():
    test_sync_invoke()         # 先跑同步
    await test_async_ainvoke() # 再跑异步

await main()

典型输出：

=== 同步 invoke ===
  [同步] 正在发送第 1 个请求...
  [同步] 正在发送第 2 个请求...
  ...
总耗时: 8.73 秒

=== 异步 ainvoke ===
  [异步] 瞬间派发 5 个请求...
总耗时: 1.92 秒

5个请求，同步耗时 ~9秒，异步耗时 ~2秒——快了 4-5 倍。请求越多，差距越大。

关键知识点：ainvoke() 和 asyncio.gather() 各自干了什么？

看完对比你可能会问：既然 ainvoke() 是异步函数，为什么逐个 await ainvoke() 跟同步一样慢？

因为 ainvoke() 和 gather() 解决的是两个不同的问题：

ainvoke() 解决的是 → "等待时不阻塞"（让出 CPU，别的任务有机会插进来）
gather()  解决的是 → "同时派发多个任务"（把多个协程塞进事件循环并行跑）

回到奶茶店的比喻：

invoke() = 你站在柜台前死等，奶茶没做好之前你哪儿也去不了，后面的人也点不了单
ainvoke() = 你扫码下单后去旁边坐着，不占柜台了，后面的人可以继续点单
asyncio.gather() = 同时帮5个人下单，让奶茶店并行制作

所以：如果只有你一个人买奶茶，ainvoke()（坐着等）和 invoke()（站着等）时间一样长——因为没有"后面的人"需要你让位。ainvoke() 的价值在于"让出控制权"，而 gather() 的价值在于"利用让出的控制权塞入更多任务"。两者缺一不可。

python

# ❌ 错误理解：用了 ainvoke 就会快
# 实际效果：还是串行，因为每次 await 都在等当前这个完成
async def wrong_way():
    r1 = await llm.ainvoke("问题1")    # 等第1个完成（2秒）
    r2 = await llm.ainvoke("问题2")    # 再等第2个完成（2秒）
    r3 = await llm.ainvoke("问题3")    # 再等第3个完成（2秒）
    # 总耗时：~6秒（串行）

# ✅ 正确写法：ainvoke 负责"能让出"，gather 负责"同时跑"
async def right_way():
    tasks = [
        llm.ainvoke("问题1"),       # 创建协程，但不等待
        llm.ainvoke("问题2"),       # 创建协程，但不等待
        llm.ainvoke("问题3"),       # 创建协程，但不等待
    ]
    r1, r2, r3 = await asyncio.gather(*tasks)  # 三个请求同时发出，同时等待
    # 总耗时：~2秒（并行）

写法	效果	类比
`invoke()` 逐个调用	串行，阻塞	站在柜台前死等，一杯一杯买
`await ainvoke()` 逐个调用	串行，不阻塞但没利用起来	扫码后坐着等，但只点了1杯，没人需要你让位
`asyncio.gather(*tasks)`	并行，耗时 ≈ 最慢的那个	同时下单5杯，谁好了取谁

*tasks 是 Python 的解包语法：gather(*[a, b, c]) 等价于 gather(a, b, c)。

ainvoke是一个 异步协程（async coroutine），在列表推导式中直接调用它 不会真正执行异步请求，只会返回一堆未执行的协程对象（coroutine objects）

以上异步批量处理可以直接调用langchain内部abatch

python

from langchain_openai import ChatOpenAI

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

# 准备 5 个测试问题
prompts = [
    "用一句话介绍一下北京",
    "用一句话介绍一下上海",
    "用一句话介绍一下广州",
    "用一句话介绍一下深圳",
    "用一句话介绍一下杭州"
]

results = await llm.abatch(prompts)

运行环境差异：Jupyter vs 普通 .py 文件

你可能注意到了，上面的代码直接写了 await main()，而不是 asyncio.run(main())。这是因为运行环境不同：

环境	启动异步的方式	原因
Jupyter Notebook / IPython	`await main()`	Jupyter 内部已经有一个事件循环在运行，不能再创建新的
普通 .py 文件	`asyncio.run(main())`	需要自己创建并启动事件循环

python

# ===== Jupyter Notebook 中 =====
async def main():
    response = await llm.ainvoke("你好")
    print(response.content)

await main()           # ✅ 直接 await
# asyncio.run(main())  # ❌ 会报错：Cannot run nested event loops


# ===== 普通 .py 文件中 =====
import asyncio

async def main():
    response = await llm.ainvoke("你好")
    print(response.content)

# await main()          # ❌ 会报错：await 只能在 async 函数内使用
asyncio.run(main())     # ✅ 创建事件循环并运行

本课程代码默认在 Jupyter 环境中运行，所以统一使用 await 写法。如果你在 PyCharm 的 .py 文件中运行，把 await main() 改成 asyncio.run(main()) 即可。

适用场景：

需要同时处理多个请求（批量调用、并行对比）
Web 服务、API 接口（FastAPI 等异步框架）
对响应时间有要求的应用

3.3 流式调用 - `stream()`（打字机效果）

实现打字机效果，提升用户体验：

python

def streaming_example():
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")

    print("AI回答: ")
    full_message = None
    for chunk in llm.stream("请写一首关于春天的诗"):
        # 累积消息块
        full_message = chunk if full_message is None else full_message + chunk
        print(chunk.content, end="", flush=True)

    # 完整消息
    print(f"\n\n完整消息:\n{full_message.content}")

streaming_example()

flush=True 的作用是强制将内存缓冲区中的内容立刻推送到屏幕上显示，而不是等攒够了一定数量或者遇到换行符才显示。
end=""：取消print默认的\n换行符，让输出结果在同一行拼接。

流式事件监听（高级用法）：

python

async def stream_events():
    async for event in llm.astream_events("你好"):
        if event["event"] == "on_chat_model_start":
            print(f"输入: {event['data']['input']}")
        elif event["event"] == "on_chat_model_stream":
            print(f"Token: {event['data']['chunk'].content}", end="",flush=True)
        elif event["event"] == "on_chat_model_end":
            print(f"\n完成!")
await stream_events()

适用场景：

聊天机器人、对话系统
长文本生成（让用户看到进度）
实时交互应用

3.4 批次调用 - `batch()`（并行处理）

并行处理多个独立请求：

python

def batch_example():
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")

    questions = [
        "什么是Python？",
        "什么是JavaScript？",
        "什么是Go语言？"
    ]

    responses = llm.batch(questions)
    for q, r in zip(questions, responses):
        print(f"Q: {q}")
        print(f"A: {r.content}\n")

batch_example()

适用场景：

批量处理多个独立请求
数据分析、批量内容生成
不需要按顺序返回结果

批量异步调用：

python

async def batch_async():
    questions = [
        "什么是LangChain？",
        "LangChain的核心组件有哪些？",
        "如何使用LangChain构建Agent？"
    ]
    responses = await llm.abatch(questions)
    for q, r in zip(questions, responses):
        print(f"Q: {q}\nA: {r.content}\n")

await batch_async()

维度	`batch` (同步)	`abatch` (异步)
并发模型	多线程 (`ThreadPoolExecutor`)	单线程事件循环 (`asyncio`)
工作原理	开 N 个线程，每个线程发一个请求并阻塞等待响应	1 个线程内注册 N 个 IO 任务，发完请求后挂起，哪个响应回来了就处理哪个
CPU 占用	高（线程创建/切换/销毁有开销）	极低（只是内存中的回调注册）
内存占用	高（每个线程默认占 ~8MB 栈空间）	低（每个协程仅占几 KB）
GIL 影响	⚠️ 受 GIL 限制，无法真正并行 CPU 计算	✅ 不受 GIL 影响（因为本来就是单线程）
LLM API 调用瓶颈	线程数上限 = 并发上限	几乎无上限（受限于文件描述符/连接池）

4、调用配置与高级特性

4.1 运行时配置

通过 config 参数传递运行时配置：

python

from langchain_openai import ChatOpenAI

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

response = llm.invoke(
    "讲一个笑话",
    config={
        "tags": ["humor", "demo"],          # 标签
        "metadata": {"user_id": "123"},     # 元数据
    }
)
print(response)

config 字典是在 LangChain 框架层流转的"元数据"，专门用于系统的观测、调试和日志记录，它与业务层面的问答结果 response 是严格分离的。

用途：

调试和追踪
日志记录（通过 metadata 传递额外信息）
监控和分析

. 这些 config 数据到底去哪了？

这些数据被 LangChain 的 回调系统（Callback System） 拦截并收集起来了。它们主要用于以下两个场景：

场景 A：云端监控与可视化（LangSmith）通过在环境变量中配置了 LangSmith（LangChain 官方的监控平台）

场景 B：本地回调拦截,通过回调处理器（Callback Handler）来拦截它们

python

from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler

# 1. 自定义一个回调处理器，拦截并打印运行信息
class MyDebugCallback(BaseCallbackHandler):
    def on_chat_model_start(self, serialized, prompts, **kwargs):
        print("\n" + "="*40)
        print(" [拦截到 LLM 启动请求]")
        print(f" Tags: {kwargs.get('tags')}")
        print(f" Metadata: {kwargs.get('metadata')}")

        print("="*40 + "\n")

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

# 2. 在调用时，将你的回调处理器传进去
response = llm.invoke(
    "讲一个短笑话",
    config={
        "tags": ["humor", "demo"],
        "metadata": {"user_id": "123"},
        "callbacks": [MyDebugCallback()]  # 关键：挂载你的回调
    }
)

print("最终返回的 response 依然只有大模型的内容：")
print(response.content)

messages / prompts：包裹里的货品（比如"讲个笑话"）。
config / kwargs：运单上的加急标签、客户编号（比如 tags, metadata）。
serialized：这辆快递车的车辆行驶证。上面写着车牌号（模型名称）、排量（Temperature）、车辆品牌（OpenAI 类路径），它不影响包裹的内容，但对车队管理员（框架和开发者）来说是必不可少的档案。

4.2 运行时动态切换模型

2.4 节介绍了用 init_chat_model 在初始化时选择不同模型。这里展示一个更高级的用法——不重新初始化，在调用时通过 config 参数动态切换：

python

from langchain.chat_models import init_chat_model

configurable_model = init_chat_model(temperature=0)

# 使用GPT-4
result1 = configurable_model.invoke(
    "你好",
    config={"configurable": {"model": "gpt-4o-mini"}}
)

# 使用Claude
result2 = configurable_model.invoke(
    "你好",
    config={"configurable": {"model": "claude-sonnet-4-6"}}
)

适用场景：

需要在运行时根据用户选择切换模型
A/B 测试不同模型效果
多租户系统（不同客户使用不同模型）

消息类型与调用方式 ​

1、消息类型：传什么 ​

2、传入方式：怎么传 ​

快速决策表 ​

方式一：直接传入字符串（最简单） ​

方式二：传入消息列表（最常用） ​

方式三：使用元组或字典（最灵活） ​

3、调用方式：怎么调 ​

3.1 同步调用 - invoke()（最常用） ​

3.2 异步调用 - ainvoke()（高并发） ​

前置知识：同步 vs 异步 ​

基础用法 ​

invoke() vs ainvoke() 性能对比 ​

关键知识点：ainvoke() 和 asyncio.gather() 各自干了什么？ ​

运行环境差异：Jupyter vs 普通 .py 文件 ​

3.3 流式调用 - stream()（打字机效果） ​

3.4 批次调用 - batch()（并行处理） ​

4、调用配置与高级特性 ​

4.1 运行时配置 ​

4.2 运行时动态切换模型 ​

消息类型与调用方式

1、消息类型：传什么

2、传入方式：怎么传

快速决策表

方式一：直接传入字符串（最简单）

方式二：传入消息列表（最常用）

方式三：使用元组或字典（最灵活）

3、调用方式：怎么调

3.1 同步调用 - `invoke()`（最常用）

3.2 异步调用 - `ainvoke()`（高并发）

前置知识：同步 vs 异步

基础用法

invoke() vs ainvoke() 性能对比

关键知识点：ainvoke() 和 asyncio.gather() 各自干了什么？

运行环境差异：Jupyter vs 普通 .py 文件

3.3 流式调用 - `stream()`（打字机效果）

3.4 批次调用 - `batch()`（并行处理）

4、调用配置与高级特性

4.1 运行时配置

4.2 运行时动态切换模型