一、历史对话管理

1）规则：大模型是无状态的，每次调用都需要传递完整对话历史。

# 第一轮
[system , user] -> AI回复 -> 保存回复

# 第二轮
[system , user, assistant, user] -> AI回复 -> 保存回复

# 第三轮
[system , user, assistant, user, assistant, user] -> AI回复 -> 保存回复

2）调用举例

import os

from rich import print as rprint
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("NEB_API")
DEEPSEEK_BASE_URL = os.getenv("NEB_URL")

model = ChatOpenAI(
  model="nebulacoder-cot-v8.0",
  api_key=DEEPSEEK_API_KEY,
  base_url=DEEPSEEK_BASE_URL,
)

conversation = []

# 第一次
conversation.append({"role": "user", "content": "我叫张三"})
response1 = model.invoke(conversation)

# 保存AI回复
conversation.append({"role": "assistant", "content": response1.content})

# 第二次 传递完整历史
conversation.append({"role": "user", "content": "我叫什么？"})
response2 = model.invoke(conversation)
rprint(response2.content)

# AI回复：你叫张三。这是你在我们对话开始时告诉我的名字。

3）对话历史优化

问题：对话历史会越来越长，消耗大量token成本。

方案在之前大模型基础中提到过，可以采用滑动窗口，摘要压缩等方式。

这里以滑动窗口举例。

* 总是保留system消息（定义角色）。

* 只保留最近N轮，丢弃更早的对话。

import os

from rich import print as rprint
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("NEB_API")
DEEPSEEK_BASE_URL = os.getenv("NEB_URL")

model = ChatOpenAI(
    model="nebulacoder-cot-v8.0",
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)

def keep_recent_messages(messages, max_pairs = 3):
    # 分离system消息和
    system_messages = [m for m in messages if m.get("role") == "system"]
    conversation_messages = [m for m in messages if m.get("role") != "system"]

    # 只保留最近的消息对
    recent_messages = conversation_messages[-(max_pairs * 2):]

    return system_messages + recent_messages



conversation = [
    {"role": "system", "content": "你是一个python导师。"}
]

# 第一次
conversation.append({"role": "user", "content": "什么是列表？用一句话解释。"})
response1 = model.invoke(conversation)
conversation.append({"role": "assistant", "content": response1.content})
# 第二次
conversation.append({"role": "user", "content": "列表和元组有什么区别？用一句话解释。"})
response2 = model.invoke(conversation)
conversation.append({"role": "assistant", "content": response2.content})
# 第三次
conversation.append({"role": "user", "content": "什么是字典？用一句话解释。"})
response3 = model.invoke(conversation)
conversation.append({"role": "assistant", "content": response3.content})

rprint(f"原始消息数：{len(conversation)}")

# 优化：只保留最近N轮
optimized = keep_recent_messages(conversation, max_pairs=2)

# 添加新的用户问题
optimized.append({"role": "user", "content": "我问的第一个问题是什么？"})
response4 = model.invoke(optimized)

rprint(f"AI回复：{response4.content}")

二、多轮对话聊天机器人案例

import os

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv(override=True)

DEEPSEEK_API_KEY = os.getenv("NEB_API")
DEEPSEEK_BASE_URL = os.getenv("NEB_URL")

model = ChatOpenAI(
    model="nebulacoder-cot-v8.0",
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
)


def keep_recent_messages(messages, max_pairs=3):
    # 分离system消息和
    system_messages = [m for m in messages if m.get("role") == "system"]
    conversation_messages = [m for m in messages if m.get("role") != "system"]

    # 只保留最近的消息对
    recent_messages = conversation_messages[-(max_pairs * 2):]

    return system_messages + recent_messages


# 维护一个消息列表
conversation = [
    {"role": "system", "content": "你是一个友好的AI聊天助手，你将扮演刘亦菲的角色，回答用户的问题。"}
]

EXIT_WORD = "exit"
PAIRS = 10

print(f"请输入问题，当输入{EXIT_WORD}时结束对话")

i = 1
while True:
    print("\n", "=" * 10, f"第{i}轮对话", "\n")

    user_input = input("请输入：")

    if user_input == EXIT_WORD:
        print("会话结束。")
        break

    conversation.append({"role": "user", "content": user_input})

    # 拼接AI回复信息
    reply_content = ""

    # 优化历史记忆
    memory_message = keep_recent_messages(conversation, max_pairs=PAIRS)

    for chunk in model.stream(memory_message):
        print(chunk.content, end="", flush=True)
        reply_content += chunk.content

    print("\n", "=" * 10, f"第{i}轮对话结束", "\n")
    i += 1
    
    conversation.append({"role": "assistant", "content": reply_content})