001.LangChain 核心抽象：Models 与 Messages

原作者：怕浪猫

LangChain 最基础、最核心的两个抽象概念——Models（模型）和 Messages（消息）：Models 是 LangChain 与大模型交互的“入口”，决定了回答的质量和能力；Messages 是交互的“载体”，定义了用户、AI 和系统之间的对话逻辑。

1. LLM 与 ChatModel 的区别与适用场景

LangChain 中，与大模型交互的核心接口分为两类：LLM（大语言模型）和 ChatModel（聊天模型）。很多新手容易混淆两者，其实它们的核心区别在于“交互格式”和“适用场景”，选择对了模型类型，能大幅提升开发效率。

1. 核心区别（一张表看懂）

对比维度	LLM（大语言模型）	ChatModel（聊天模型）
交互格式	纯文本输入（字符串），输出纯文本（字符串）	消息列表输入（Message 对象），输出消息对象（AIMessage）
核心特点	简单直接，无需关注消息结构，适合纯文本生成	支持多轮对话、系统提示，结构清晰，适合聊天场景
调用方式	使用 `invoke(text)` 调用，输入字符串	使用 `invoke(messages)` 调用，输入消息列表
典型模型	GPT-3（text-davinci-003）、Llama 2（text 模型）	GPT-3.5-turbo、GPT-4、Claude、Llama 3（chat 模型）

2. 代码示例（直观对比）

1. LLM 调用示例（以 OpenAI 的 text-davinci-003 为例）

from langchain_openai import OpenAI
from dotenv import load_dotenv
import os

load_dotenv()

# 初始化LLM（纯文本交互）
llm = OpenAI(
    model_name="text-davinci-003",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7
)

# 调用：输入纯文本字符串
response = llm.invoke("简单介绍LangChain")
print("LLM输出（纯文本）：", response)

2. ChatModel 调用示例（以 GPT-3.5-turbo 为例）

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

load_dotenv()

# 初始化ChatModel（消息列表交互）
chat_model = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7
)

# 调用：输入消息列表（此处为单条用户消息）
response = chat_model.invoke([HumanMessage(content="简单介绍LangChain")])
print("ChatModel输出（消息对象）：", response.content)

3. 适用场景选择（新手必看）

优先选 ChatModel 的场景：多轮对话（如聊天机器人）、需要系统提示（System Prompt）、需要明确区分用户/AI 消息、使用主流聊天模型（GPT-3.5/4、Claude、Llama 3）；
选 LLM 的场景：纯文本生成（如文案、摘要）、使用旧版文本模型（如 text-davinci-003）、简单场景无需复杂消息结构；
注意：LangChain 1.0+版本更推荐使用 ChatModel，其功能更全面、更贴合当前大模型的交互逻辑，LLM 更多用于兼容旧版模型。

3. 消息类型详解：HumanMessage、AIMessage、SystemMessage

在 ChatModel 交互中，所有输入都是“消息列表”，LangChain 定义了 3 种核心消息类型，分别对应“用户输入、AI 输出、系统提示”，它们共同构成了多轮对话的逻辑，掌握这些消息类型，才能灵活实现复杂的聊天场景。

1. 核心消息类型（3 种必掌握）

所有消息类型均继承自 BaseMessage，核心区别在于“角色”和“用途”，具体如下：

1. SystemMessage（系统消息）

作用：定义 AI 的“角色、行为准则、回答要求”，相当于给 AI 设定“人设”，贯穿整个对话过程；
特点：通常放在消息列表的最前面，只需要设置一次（多轮对话中可重复设置，覆盖之前的准则）；
示例：SystemMessage(content="你是一名LangChain开发专家，回答需包含代码示例，简洁明了。")。

2. HumanMessage（用户消息）

作用：用户输入的查询、问题或指令，是 AI 生成回答的核心依据；
特点：多轮对话中，每次用户输入都是一条 HumanMessage，与 AI 的 AIMessage 交替出现；
示例：HumanMessage(content="如何使用LangChain的SystemMessage？")。

3. AIMessage（AI 消息）

作用：AI 生成的回答，通常是 ChatModel 的输出结果；
特点：多轮对话中，需将历史 AIMessage 加入消息列表，让 AI“记住”之前的回答；
示例：AIMessage(content="使用SystemMessage需导入对应的类，放在消息列表最前面，示例如下：...")。

2. 代码示例（多轮对话，完整消息流程）

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

load_dotenv()

chat_model = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.6
)

# 构建多轮对话消息列表（系统消息→用户消息→AI消息→新用户消息）
messages = [
    # 系统消息：设定AI角色和回答要求
    SystemMessage(content="你是LangChain消息类型助手，回答仅围绕3种核心消息类型，不扩展其他内容。"),
    # 第一轮：用户提问，AI回答
    HumanMessage(content="LangChain有哪几种核心消息类型？"),
    AIMessage(content="核心消息类型有3种：SystemMessage（系统提示）、HumanMessage（用户输入）、AIMessage（AI输出）。"),
    # 第二轮：用户追问，AI继续回答
    HumanMessage(content="SystemMessage的作用是什么？")
]

# 调用模型，获取新的AI消息
new_ai_message = chat_model.invoke(messages)
print("新AI回答：", new_ai_message.content)

# 将新的AI消息加入列表，用于下一轮对话
messages.append(new_ai_message)
print("\n完整消息列表：")
for msg in messages:
    print(f"【{msg.type}】: {msg.content}")

3. 其他常用消息类型（可选）

除了 3 种核心消息类型，LangChain 还提供两种辅助消息类型，适合特殊场景：

FunctionMessage：用于工具调用场景，存储工具调用的结果（后续 Agent 章节详细讲解）；
ToolMessage：与 FunctionMessage 配套，传递工具调用的参数和结果。

4. 控制生成行为：temperature、max_tokens、stop sequences

调用大模型时，我们需要控制其生成行为（如回答的随机性、长度、结束条件），避免出现“回答过长、偏离主题、随机性过高”等问题。LangChain 支持通过 3 个核心参数控制生成行为，所有模型（ChatModel/LLM）均适用。

1. 核心参数详解（3 个必掌握）

参数名称	作用	取值范围	使用建议
temperature（温度）	控制回答的随机性和创造性，值越高，回答越随机、有创造性；值越低，回答越严谨、固定	0 ~ 2	严谨场景（如问答、代码）：0.1~~0.5；创意场景（如文案、故事）：0.7~~1.5
max_tokens（最大 tokens 数）	控制 AI 回答的最大长度（tokens 是模型处理文本的基本单位，1 个中文约等于 1~2 个 tokens）	正整数（根据模型限制调整）	根据需求设定，避免回答过长（如聊天场景设 100~~200，长文本场景设 500~~1000）
stop sequences（停止序列）	设定 AI 回答的“终止条件”，当 AI 生成的文本包含停止序列时，立即停止生成	字符串或字符串列表	适合固定格式场景（如生成列表时，以“### 结束”为停止序列）

2. 代码示例（参数实战）

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

load_dotenv()

# 1. 低temperature（0.2）：严谨、固定回答
chat_low_temp = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.2,
    max_tokens=100
)

# 2. 高temperature（1.5）：随机、有创造性回答
chat_high_temp = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=1.5,
    max_tokens=100
)

# 3. 带stop sequences：生成到“### 结束”时停止
chat_stop_seq = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    max_tokens=200,
    stop=["### 结束"]  # 停止序列
)

# 测试不同参数的效果
question = "用3句话介绍LangChain的Models抽象"

print("=== 低temperature（0.2） ===")
print(chat_low_temp.invoke([HumanMessage(content=question)]).content)

print("\n=== 高temperature（1.5） ===")
print(chat_high_temp.invoke([HumanMessage(content=question)]).content)

print("\n=== 带stop sequences ===")
print(chat_stop_seq.invoke([HumanMessage(content=question + "，结尾加上### 结束")]).content)

3. 注意事项

max_tokens 设定不宜过大，否则会增加成本（云端模型按 tokens 收费），且可能导致回答冗余；
stop sequences 需根据生成格式设定，避免出现“未生成完整内容就停止”的情况；
不同模型对参数的支持略有差异（如部分本地模型不支持 stop sequences），需参考对应模型的文档。

5. 流式输出（Streaming）实现与前端对接

默认情况下，LangChain 调用模型时，会等待模型生成完整回答后再返回（同步输出），这种方式在回答较长时，会出现“长时间无响应”的问题，影响用户体验。流式输出（Streaming）可实现“边生成、边返回”，类似 ChatGPT 的打字效果，是聊天类应用的必备功能。

1. 流式输出核心原理

流式输出通过“迭代器”实现：模型生成文本时，会将内容分块返回，LangChain 通过 stream() 方法返回迭代器，开发者可遍历迭代器，逐块获取生成内容，实现实时输出。

2. LangChain 流式输出代码示例（后端）

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

load_dotenv()

# 初始化ChatModel，支持流式输出
chat_stream = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7,
    streaming=True  # 开启流式输出
)

# 流式调用：使用stream()方法，返回迭代器
print("流式输出（边生成边显示）：")
for chunk in chat_stream.stream([HumanMessage(content="详细介绍LangChain的流式输出原理，分3点说明。")]):
    # 逐块打印生成内容，不换行
    print(chunk.content, end="", flush=True)

6. 异步调用（async/await）提升性能

默认情况下，LangChain 调用模型是“同步调用”——一次只能处理一个请求，后续请求需等待前一个请求完成，效率较低。异步调用（async/await）可实现“同时处理多个请求”，提升并发性能，适合高并发场景（如多用户同时提问）。

LangChain 的所有模型接口均支持异步调用，只需将 invoke() 替换为 ainvoke()，配合 async/await 语法即可实现。

1. 异步调用基础示例

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

load_dotenv()

# 初始化ChatModel（支持异步调用）
chat_async = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7
)

# 定义异步函数
async def async_chat(question):
    # 异步调用：ainvoke()
    response = await chat_async.ainvoke([HumanMessage(content=question)])
    return response.content

# 运行异步函数
if __name__ == "__main__":
    question = "LangChain异步调用的优势是什么？"
    result = asyncio.run(async_chat(question))
    print("异步调用结果：", result)

2. 多请求并发异步调用（核心实战）

异步调用的核心优势是“并发处理多个请求”，下面示例实现“同时处理 3 个请求”，对比同步调用和异步调用的效率差异：

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

load_dotenv()

chat_async = ChatOpenAI(
    model_name="gpt-3.5-turbo",
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.7
)

# 1. 同步调用（依次处理3个请求）
def sync_chat(questions):
    start_time = time.time()
    results = []
    for q in questions:
        response = chat_async.invoke([HumanMessage(content=q)])
        results.append(response.content)
    end_time = time.time()
    print(f"同步调用耗时：{end_time - start_time:.2f}秒")
    return results

# 2. 异步调用（并发处理3个请求）
async def async_chat_single(question):
    return await chat_async.ainvoke([HumanMessage(content=question)])

async def async_chat_batch(questions):
    start_time = time.time()
    # 并发执行多个异步任务
    tasks = [async_chat_single(q) for q in questions]
    results = await asyncio.gather(*tasks)
    end_time = time.time()
    print(f"异步调用耗时：{end_time - start_time:.2f}秒")
    return [res.content for res in results]

# 测试对比
if __name__ == "__main__":
    questions = [
        "介绍LangChain的Models抽象",
        "介绍LangChain的Messages类型",
        "介绍LangChain的流式输出"
    ]

    # 同步调用
    sync_results = sync_chat(questions)
    # 异步调用
    async_results = asyncio.run(async_chat_batch(questions))

    print("\n同步调用结果：", sync_results)
    print("\n异步调用结果：", async_results)

运行结果说明：异步调用耗时约为同步调用的 1/3（具体耗时取决于网络和模型响应速度），并发优势明显；代码中使用 asyncio.gather() 实现多个异步任务的并发执行，是高并发场景的常用方法。

3. 注意事项

异步调用需配合支持异步的框架（如 FastAPI、Starlette），Flask 默认不支持异步，需使用 Flask-AsyncExt 扩展；
云端模型有 API 调用频率限制，并发请求数量不宜过多，避免触发限流；
本地模型的异步调用效果取决于模型部署工具（如 vLLM 支持异步，Ollama 异步支持有限）。