🎯 课程主题

基于此前对话历史优化知识，综合运用消息列表管理、流式输出和历史优化，构建一个完整的多轮对话聊天机器人。

📝 核心知识点

1. 消息列表的维护

  • 概念说明：使用列表 messages 存储全部对话历史，初始包含 SystemMessage 设定角色。
  • 关键细节：使用字典格式（JSON 风格）描述消息，role 支持 system、user、assistant。

2. 循环交互结构

  • 概念说明：使用 while True 实现持续对话，通过输入特定退出词（如 quit）配合 break 退出循环。
  • 关键细节：需维护轮次计数器 i，每轮对话结束自增。

3. 流的式输出（Streaming）

  • 概念说明：将 model.invoke() 替换为 model.stream() 实现逐字输出，提升交互体验。
  • 关键细节：stream() 返回生成器，需用 for chunk in model.stream(messages) 循环接收；每个 chunk 的 content 片段需拼接为完整回复字符串，再以字典形式追加到消息列表。

4. 历史优化集成

  • 概念说明：在每次调用模型前，调用此前编写的 optimize_memory 函数，裁剪消息列表至最近 N 轮。
  • 关键细节：优化后的列表用于模型调用，但模型回复需追加到原始的 messages 列表中（而非优化后的临时列表），否则对话历史会丢失。

🏗️ 架构与工作流

     ┌─────────────┐
     │ 初始化消息列表 │  [SystemMessage, ...]
     └──────┬──────┘
            ↓
     ┌─────────────┐
     │  while True  │ ← 进入循环
     └──────┬──────┘
            ↓
     ┌─────────────┐
     │ 用户输入问题  │  input()
     └──────┬──────┘
            ↓
     ┌─────────────┐
     │ 判断是否退出  │  if user_input == "quit" → break
     └──────┬──────┘
            ↓
     ┌─────────────────────┐
     │ 用户消息追加到列表    │  messages.append({"role":"user", ...})
     └──────┬──────────────┘
            ↓
     ┌─────────────────────┐
     │ 调用 optimize_memory │  历史裁剪
     └──────┬──────────────┘
            ↓
     ┌─────────────────────┐
     │  model.stream() 流式 │  逐块输出回复
     └──────┬──────────────┘
            ↓
     ┌─────────────────────┐
     │ 拼接完整AI回复字符串 │  reply_content += chunk.content
     └──────┬──────────────┘
            ↓
     ┌─────────────────────────┐
     │ AI回复追加到原始messages │  messages.append({"role":"assistant", ...})
     └──────┬──────────────────┘
            ↓
     ┌─────────────────┐
     │ 轮次 i += 1      │
     └────────┬────────┘
            ↓（回到循环）

💻 代码实战

from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

model = ChatOpenAI(
    model="gpt-4o",
    base_url="https://api.claude.ai/v1",
    api_key="your-api-key"
)

EXIT_WORD = "quit"
MAX_PAIRS = 10

messages = [
    SystemMessage(content="你是小谷姐姐，是尚硅谷教育的数字员工，也是一名耐心友好的AI助手，可以回答学生的问题。")
]

def optimize_memory(messages, max_pairs=3):
    system_messages = [m for m in messages if isinstance(m, SystemMessage)]
    conversation_messages = [m for m in messages if not isinstance(m, SystemMessage)]
    recent_messages = conversation_messages[-(max_pairs * 2):]
    return system_messages + recent_messages


print("=" * 10)
print(f"请输入具体的问题（输入 {EXIT_WORD} 结束对话）")

i = 1
while True:
    print("=" * 10)
    print(f"第{i}轮对话开始")
    print()

    user_input = input("请输入: ")

    if user_input == EXIT_WORD:
        print("会话已结束，欢迎下次再来！")
        break

    messages.append(HumanMessage(content=user_input))

    memory_messages = optimize_memory(messages, max_pairs=MAX_PAIRS)

    print("小谷姐姐: ", end="", flush=True)
    reply_content = ""
    for chunk in model.stream(memory_messages):
        if chunk.content:
            print(chunk.content, end="", flush=True)
            reply_content += chunk.content
    print()

    messages.append(AIMessage(content=reply_content))

    print(f"第{i}轮对话结束")
    print()
    i += 1

⚠️ 常见问题与避坑指南

  • 模型回复需追加到原始 messages 而非 memory_messages：memory_messages 是裁剪后的临时列表，追加到它上面会在下一轮循环中丢失数据。一定要用 messages.append(AIMessage(...))。
  • 流式输出时 end="" 和 flush=True：避免 print 自动换行并确保实时输出，end="" 让每个 chunk 拼接到同一行，flush=True 立即刷新缓冲区。
  • 历史优化函数需先定义再使用：若在 Notebook 中分 cell 执行，确保 optimize_memory 函数所在 cell 已运行再执行主逻辑。
  • 模型的流式输出不会自动拼接：stream() 返回片段，必须手动拼接成完整字符串再存入消息列表，否则后期无法正确回传历史。

💡 个人总结与延伸

本节完成了从"理论认知"到"完整应用"的跨越——将模型调用、消息管理、历史优化、流式输出组合成一个可直接运行的聊天机器人。这种模式是构建生产级 ChatBot 的基础骨架，后续可直接在此基础上扩展工具调用、RAG 检索增强等功能。实际项目中建议将历史优化和消息管理封装为独立的 Memory 类，便于复用和维护。

最近更新:
Contributors: yangwenguang