LangChain框架对接阿里云通义千言大模型 的实践合集，完整实现了「无状态对话、带上下文记忆对话、超长对话修剪、流式实时输出」四种生产级别的对话能力，从基础到进阶。

一、前置知识 & 整体导入说明

✅ 核心依赖包导入作用

from langchain_community.chat_models import ChatTongyi  # 通义千言 Chat 模型
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import trim_messages

1. ChatTongyi：LangChain封装的通义千言专用对话模型类，是对接阿里云通义大模型的核心入口，支持qwen-turbo/qwen-plus/qwen-max等所有通义系列模型
2. 三种核心消息类：
   • HumanMessage：用户发送的消息（核心）
   • AIMessage：AI模型返回的消息（自动记录）
   • SystemMessage：系统指令，定义AI的角色/回答风格（如你是友好的助手）
3. ChatPromptTemplate：对话专用的提示词模板，结构化拼接「系统消息+历史对话+新问题」，是LangChain对话的核心
4. MessagesPlaceholder：历史对话占位符，用于动态填充对话上下文，是实现上下文记忆的关键
5. RunnableWithMessageHistory：给普通的LangChain链 赋予「会话记忆」能力 的包装器，核心功能
6. InMemoryChatMessageHistory：内存级别的对话历史存储器，会话数据存在内存中，程序重启丢失，适合测试/轻量场景
7. trim_messages：对话历史修剪工具，解决「上下文过长触发模型token限制」的核心方案

✅ 密钥配置 & 路径处理

import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from config import DASHSCOPE_API_KEY

1. 通义千言的调用需要阿里云的 DashScope API密钥，必须从config.py导入，这是官方规范（密钥不硬编码，避免泄露）
2. sys.path.append(...)：将当前文件所在目录加入Python路径，确保能成功导入同目录的config.py

---

二、核心知识点铺垫（重中之重）

在看具体函数前，必须先搞懂2个核心概念，否则代码逻辑会看不懂：

✅ 核心概念1：LangChain的「无状态」特性

所有大模型本身都是「无状态」的，包括通义千言。

• 模型不会主动记住你上一轮问了什么
• 每一次llm.invoke()调用，都是一次独立的请求，模型只能看到本次传入的消息内容
• 所谓的「上下文对话」，本质是：把「历史对话记录+本次新问题」一起打包传给模型，模型通过读取历史记录实现「记忆」

✅ 核心概念2：会话隔离（session_id）

代码中所有带记忆的对话都用到了session_id（如user_001/user_002），作用是：

• 不同用户/不同会话用不同的session_id
• 每个session_id对应独立的对话历史，互不干扰（比如小明的对话记录，小红看不到）
• 核心载体是全局字典store = {}，key=session_id，value=该会话的完整历史消息

---

三、逐模块解析4种对话模式

===================== 1. 基础无状态对话（模型不记忆历史） stateless_chat()

✅ 核心功能

演示「原生的无记忆对话」，这是大模型的默认形态，每一轮对话完全独立。

✅ 代码详解

def stateless_chat():
    # 初始化通义千言模型，指定基础型号
    llm = ChatTongyi(model="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)
    # 第一轮对话：用户发送消息（注意：这里你代码里笔误了，应该是「你好，我叫小明，今天天气如何」）
    msg1 = [HumanMessage(content="你好，我叫今天天气如何")]
    res1 = llm.invoke(msg1) # 单次独立调用
    print(f"无状态对话-第一轮: {res1.content}")

    # 第二轮对话：问「我叫什么名字？」
    msg2 = [HumanMessage(content="我叫什么名字？")]
    res2 = llm.invoke(msg2) # 新的独立调用，无任何历史信息
    print(f"无状态对话-第二轮: {res2.content}")

✅ 关键结论

运行结果一定是：模型回答「我不知道你的名字」

• 因为两次invoke()是完全独立的，msg2里只有「我叫什么名字？」这一句话
• 模型没有收到任何「你叫XX」的历史信息，自然无法回答
• 这是所有大模型的原生特性，也是后续实现「有状态对话」的原因

---

===================== 2. 带历史的对话（模型记忆上下文） stateful_chat()

✅ 核心功能

实现 带上下文记忆的多轮对话 + 多会话隔离，是业务开发中最常用的核心功能，这部分是代码的重中之重！

✅ 前置核心：会话历史存储器 & 获取函数

# 全局字典：存储所有会话的历史记录，内存级存储（程序重启丢失）
store = {}
# 根据session_id获取对应会话的历史，不存在则新建
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

• InMemoryChatMessageHistory()：每个会话都会生成一个该对象，内部维护了一个「消息列表」，按顺序存储 SystemMessage/HumanMessage/AIMessage
• 该函数是RunnableWithMessageHistory的核心依赖，作用是「告诉LangChain：去哪里取当前会话的历史」

✅ 核心步骤拆解

步骤1：初始化模型，新增temperature参数

llm = ChatTongyi(model="qwen-turbo", temperature=0.7, dashscope_api_key=DASHSCOPE_API_KEY)

• temperature：随机性控制参数，取值0~1
  • 值越小（如0.1）：回答越精准、严谨、重复率高，适合问答/知识库场景
  • 值越大（如0.9）：回答越发散、有创意，适合文案/创作场景
  • 0.7是兼顾「严谨+灵活」的通用值

步骤2：定义结构化的对话提示词模板

prompt = ChatPromptTemplate.from_messages([
    SystemMessage(content="你是一个友好的助手，回答简洁明了"), # AI的角色定义
    MessagesPlaceholder(variable_name="history"), # 历史对话占位符，动态填充
    HumanMessage(content="{input}") # 用户本次输入的占位符
])

✅ 这个模板是上下文对话的灵魂，最终传给模型的完整消息格式是：

[系统消息：你是一个友好的助手，回答简洁明了] + [历史对话：用户上一轮说的话 + AI上一轮的回答] + [用户本次输入]

• MessagesPlaceholder(variable_name="history")：动态填充对话历史，无需手动拼接，LangChain自动处理
• {input}：动态填充用户本次输入的内容

步骤3：创建基础链 + 包装「记忆能力」

# 基础链：提示词模板 拼接内容后 传给 大模型
chain = prompt | llm
# 给基础链包装「会话记忆」能力，核心！
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,  # 传入「历史获取函数」，指定历史存储位置
    input_messages_key="input", # 对应prompt中的 {input}，用户输入的key
    history_messages_key="history" # 对应prompt中的 history，历史对话的key
)

✅ RunnableWithMessageHistory的核心作用（自动完成，无需手动写逻辑）：

1. 调用时根据session_id获取该会话的历史消息
2. 将「历史消息」填充到history占位符，「用户输入」填充到input占位符
3. 调用模型得到结果后，自动把「用户输入+AI回答」追加到该会话的历史中
4. 下一轮调用时，自动读取最新的历史，形成闭环

步骤4：多会话调用演示

# 会话1：user_001，问「你好我叫小明」→「我叫什么名字」
res1 = chain_with_history.invoke({"input": "你好，我叫小明"}, config={"configurable": {"session_id": "user_001"}})
res2 = chain_with_history.invoke({"input": "我叫什么名字？"}, config={"configurable": {"session_id": "user_001"}})

# 会话2：user_002，问「你好我叫小红」→「我叫什么名字」
res3 = chain_with_history.invoke({"input": "你好，我叫小红"}, config={"configurable": {"session_id": "user_002"}})
res4 = chain_with_history.invoke({"input": "我叫什么名字？"}, config={"configurable": {"session_id": "user_002"}})

✅ 关键结论

• user_001的第二轮回答：你叫小明
• user_002的第二轮回答：你叫小红
• 两个会话的历史完全隔离，不会互相干扰，这是生产级对话的必备能力

---

===================== 3. 对话历史修剪（防止上下文过长） trimmed_chat()

✅ 核心痛点（为什么需要修剪）

1. 所有大模型都有 上下文窗口限制（比如qwen-turbo的上下文窗口是8k tokens）
2. 如果多轮对话的历史消息无限累加，最终会触发「token超限」，模型报错无法调用
3. 即使没超限，过长的历史会让模型推理变慢、回答质量下降，还会增加调用成本
   ✅ 解决方案：trim_messages 对话历史修剪，这是解决上下文过长的标准方案！

✅ 核心代码详解

步骤1：定义修剪策略

trimmer = trim_messages(
    max_tokens=200,        # 核心：历史对话的最大token数，超过则修剪
    strategy="last",       # 修剪策略：保留「最新」的消息，删除最早的
    include_system=True,   # 必选：保留系统消息，不参与修剪（AI的角色不能丢）
    token_counter=llm      # 用当前的通义模型来计算token数，精准匹配模型的token规则
)

✅ 修剪策略说明：

• strategy="last"：最常用的策略，保留最新的对话，删除最早的，符合人类对话习惯
• 修剪时只会删除历史对话，系统消息永远保留，用户本次输入永远保留
• token计算精准：用模型自身的token计算器，避免第三方计算不准导致超限

步骤2：新增「修剪链路」，组合完整链

# 组合链：先修剪历史 -> 再喂给提示词 -> 最后调用模型
chain = (
    lambda x: {
        "input": x["input"],
        "history": trimmer.invoke(x["history"]) # 核心：先修剪历史消息
    }
) | prompt | llm

✅ 执行逻辑（关键）：

1. 每次调用时，先拿到当前会话的完整历史消息
2. 用trimmer对历史消息做修剪，只保留「最新的、不超过200token」的内容
3. 把「修剪后的历史」+「用户输入」传给提示词模板
4. 最后调用模型，完美规避token超限问题

步骤3：多轮测试触发修剪

session_id = "user_003"
for i in range(5):
    res = chain_with_history.invoke(
        {"input": f"这是第 {i+1} 轮测试消息，内容较长以触发修剪：1234567890 重复多次 {str(i)*20}"},
        config={"configurable": {"session_id": session_id}}
    )

✅ 关键结论

• 前2-3轮：历史消息少，token没超200，不修剪，正常记忆
• 第4-5轮：历史消息累加，token超过200，自动修剪掉最早的1-2轮对话
• 模型依然能正常回答最新的问题，且不会报错，兼顾「记忆」和「防超限」

---

===================== 4. 流式输出（实时返回结果） stream_chat()

✅ 核心需求（为什么需要流式输出）

1. 普通的invoke()调用：模型思考完成后，一次性返回完整回答，如果回答很长，用户需要等待几秒才能看到内容，体验差
2. 流式输出：模型思考的同时，逐字/逐句实时返回回答内容，像ChatGPT网页版一样，边打字边显示，体验极佳
   ✅ 实现方式：LangChain+通义千言的流式输出，一行配置即可开启，非常简单！

✅ 核心代码详解

步骤1：初始化模型，必须开启 streaming=True

llm = ChatTongyi(model="qwen-turbo", streaming=True, temperature=0.7, dashscope_api_key=DASHSCOPE_API_KEY)

✅ 重中之重：流式输出的唯一前提是 streaming=True，没有这个参数，所有流式相关的调用都会失效！

步骤2：流式调用核心：stream() 替代 invoke()

print("流式对话输出（实时打印）：")
# 流式迭代返回结果，核心：用stream()，不是invoke()
for chunk in chain_with_history.stream(
    {"input": "请介绍一下 LangChain 的核心优势"},
    config={"configurable": {"session_id": "user_004"}}
):
    print(chunk.content, end="", flush=True) # end=""：不换行，flush=True：实时刷新打印
print()

✅ 流式核心逻辑：

1. chain.stream() 会返回一个生成器（generator），每次迭代返回一个「内容片段（chunk）」
2. 每个chunk是模型实时生成的一小段文本，逐段打印就实现了「边出边显」的效果
3. end="" 保证片段拼接在一起，flush=True 强制实时刷新控制台，避免内容缓存

✅ 关键结论

• 运行后能看到：控制台逐字实时打印回答内容，而不是等待几秒后一次性显示
• 流式输出完全兼容上下文记忆，历史对话依然有效
• 流式输出是C端产品的标配，极大提升用户体验，几乎所有对话类产品都会用

---

四、全局执行入口

✅ 执行入口

if __name__ == "__main__":
    print("===== 无状态对话 =====")
    stateless_chat()
    print("\n===== 带历史对话 =====")
    stateful_chat()
    print("\n===== 修剪对话 =====")
    trimmed_chat()
    print("\n===== 流式对话 =====")
    stream_chat()

这是Python的标准写法，当该文件被直接运行时，按顺序执行4个函数，演示所有功能。

五、核心知识点总结（必背，面试/开发都能用）

这份代码涵盖了LangChain对话开发的四大核心能力，也是90%的对话场景会用到的功能，总结如下：

1. 无状态对话：模型原生特性，llm.invoke() 单次调用，无记忆，适合简单问答
2. 有状态对话：核心是 RunnableWithMessageHistory + session_id，实现上下文记忆+多会话隔离，业务核心
3. 对话修剪：核心是 trim_messages，解决上下文过长的痛点，生产级必备
4. 流式输出：核心是 streaming=True + chain.stream()，提升用户体验，C端产品标配

✅ 扩展小知识（生产级优化）

代码中用的InMemoryChatMessageHistory是内存存储，程序重启后历史丢失，生产环境可以替换为：

• 本地存储：FileChatMessageHistory（存在文件里）
• 数据库存储：RedisChatMessageHistory/SQLChatMessageHistory（存在Redis/MySQL中）
  ✅ 替换后，所有代码逻辑完全不变，只需修改get_session_history函数的返回值，这就是LangChain的解耦优势！

完整代码

from langchain_community.chat_models import ChatTongyi  # 通义千言 Chat 模型
from langchain_core.messages import HumanMessage, SystemMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import trim_messages

# 从config.py导入API密钥
import sys
import os
# 将agenttest目录添加到Python路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from config import DASHSCOPE_API_KEY

# ===================== 1. 基础无状态对话（模型不记忆历史） =====================
def stateless_chat():
    # 初始化通义千言模型（支持 qwen-turbo/qwen-plus 等型号）
    llm = ChatTongyi(model="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)
    # 第一轮对话
    msg1 = [HumanMessage(content="你好，我叫今天天气如何")]
    res1 = llm.invoke(msg1)
    print(f"无状态对话-第一轮: {res1.content}")

    # 第二轮对话（模型不知道上一轮的名字）
    msg2 = [HumanMessage(content="我叫什么名字？")]
    res2 = llm.invoke(msg2)
    print(f"无状态对话-第二轮: {res2.content}")

# ===================== 2. 带历史的对话（模型记忆上下文） =====================
# 存储会话历史：key=session_id, value=ChatHistory
store = {}
# 获取会话历史的函数
def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

def stateful_chat():
    # 初始化通义千言模型，可指定温度（temperature）控制随机性
    llm = ChatTongyi(model="qwen-turbo", temperature=0.7, dashscope_api_key=DASHSCOPE_API_KEY)

    # 定义提示词模板：系统消息 + 历史消息 + 新消息
    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content="你是一个友好的助手，回答简洁明了"),
        MessagesPlaceholder(variable_name="history"),  # 对话历史占位符
        HumanMessage(content="{input}")  # 用户新输入
    ])
    # 组合链：提示词 + 通义模型
    chain = prompt | llm
    # 包装链，赋予历史记忆能力
    chain_with_history = RunnableWithMessageHistory(
        chain,
        get_session_history,  # 传入历史获取函数
        input_messages_key="input",
        history_messages_key="history"
    )
    # 会话1：user_001
    res1 = chain_with_history.invoke(
        {"input": "你好，我叫小明"},
        config={"configurable": {"session_id": "user_001"}}
    )
    print(f"带历史对话-会话1第一轮: {res1.content}")

    res2 = chain_with_history.invoke(
        {"input": "我叫什么名字？"},
        config={"configurable": {"session_id": "user_001"}}
    )
    print(f"带历史对话-会话1第二轮: {res2.content}")

    # 会话2：user_002（独立会话）
    res3 = chain_with_history.invoke(
        {"input": "你好，我叫小红"},
        config={"configurable": {"session_id": "user_002"}}
    )
    print(f"带历史对话-会话2第一轮: {res3.content}")

    res4 = chain_with_history.invoke(
        {"input": "我叫什么名字？"},
        config={"configurable": {"session_id": "user_002"}}
    )
    print(f"带历史对话-会话2第二轮: {res4.content}")

# ===================== 3. 对话历史修剪（防止上下文过长） =====================
def trimmed_chat():
    llm = ChatTongyi(model="qwen-turbo", dashscope_api_key=DASHSCOPE_API_KEY)
    # 定义修剪策略：最大令牌数 200，保留系统消息，保留最新内容
    trimmer = trim_messages(
        max_tokens=200,
        strategy="last",
        include_system=True,
        token_counter=llm  # 用通义模型计算令牌数
    )

    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content="你是一个友好的助手，回答简洁明了"),
        MessagesPlaceholder(variable_name="history"),
        HumanMessage(content="{input}")
    ])

    # 组合链：先修剪历史 -> 再喂给提示词 -> 最后调用模型
    chain = (
        lambda x: {
            "input": x["input"],
            "history": trimmer.invoke(x["history"])
        }
    ) | prompt | llm

    chain_with_history = RunnableWithMessageHistory(
        chain,
        get_session_history,
        input_messages_key="input",
        history_messages_key="history"
    )

    # 多轮对话触发修剪
    session_id = "user_003"
    for i in range(5):
        res = chain_with_history.invoke(
            {"input": f"这是第 {i+1} 轮测试消息，内容较长以触发修剪：1234567890 重复多次 {str(i)*20}"},
            config={"configurable": {"session_id": session_id}}
        )
        print(f"修剪对话-第{i+1}轮: {res.content}")

# ===================== 4. 流式输出（实时返回结果） =====================
def stream_chat():
    # 流式输出必须开启 streaming=True
    llm = ChatTongyi(model="qwen-turbo", streaming=True, temperature=0.7, dashscope_api_key=DASHSCOPE_API_KEY)

    prompt = ChatPromptTemplate.from_messages([
        SystemMessage(content="你是一个友好的助手，用中文详细回答"),
        MessagesPlaceholder(variable_name="history"),
        HumanMessage(content="{input}")
    ])

    chain = prompt | llm
    chain_with_history = RunnableWithMessageHistory(
        chain,
        get_session_history,
        input_messages_key="input",
        history_messages_key="history"
    )

    print("流式对话输出（实时打印）：")
    # 流式迭代返回结果
    for chunk in chain_with_history.stream(
        {"input": "请介绍一下 LangChain 的核心优势"},
        config={"configurable": {"session_id": "user_004"}}
    ):
        print(chunk.content, end="", flush=True)
    print()  # 换行

# ===================== 运行测试 =====================
if __name__ == "__main__":
    print("===== 无状态对话 =====")
    stateless_chat()
    print("\n===== 带历史对话 =====")
    stateful_chat()
    print("\n===== 修剪对话 =====")
    trimmed_chat()
    print("\n===== 流式对话 =====")
    stream_chat()