阿里云多轮对话

一、工作原理

通义千问 API 是无状态的 (Stateless)。它不会自动记录历史对话。要实现多轮对话，开发者必须在每次请求中显式地传递完整的上下文信息。

以下示例为多轮对话时 messages 的状态变化：

1.1 :第一轮对话

向messages 数组添加用户问题。

[
{"role": "user", "content": "推荐一部关于太空探索的科幻电影。"}
]

1.2:第二轮对话

向messages数组添加大模型回复内容与用户的最新提问。

[
{"role": "user", "content": "推荐一部关于太空探索的科幻电影。"},
{"role": "assistant", "content": "我推荐《xxx》，这是一部经典的科幻作品。"},
{"role": "user", "content": "这部电影的导演是谁？"}
]

二、 开始使用

import os
from openai import OpenAI
client = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
def get_response(messages):
completion = client.chat.completions.create(
model="qwen-plus",
messages=messages
)
return completion.choices[0].message.content
初始化 messages
messages = []
第 1 轮
messages.append({"role": "user", "content": "推荐一部关于太空探索的科幻电影。"})
print("第1轮")
print(f"用户：{messages[0]['content']}")
assistant_output = get_response(messages)
messages.append({"role": "assistant", "content": assistant_output})
print(f"模型：{assistant_output}\\n")
第 2 轮
messages.append({"role": "user", "content": "这部电影的导演是谁？"})
print("第2轮")
print(f"用户：{messages[-1]['content']}")
assistant_output = get_response(messages)
messages.append({"role": "assistant", "content": assistant_output})
print(f"模型：{assistant_output}\\n")

三、上下文管理与优化

在实际的多轮对话应用中，上下文管理是核心挑战之一。随着对话轮次增加，消息历史会不断增长，可能超出模型的最大上下文长度限制。

3.1 上下文长度限制

通义千问等大模型都有最大上下文长度限制（如 8K、32K tokens）。当对话历史超过这个限制时，需要采取策略来管理上下文。

3.2 滑动窗口策略

最简单的策略是使用固定大小的滑动窗口，只保留最近的 N 条消息：

def sliding_window(messages, window_size=10):
"""保留最近的 N 条消息"""
if len(messages) <= window_size:
return messages
return messages[-window_size:]

3.3 历史消息压缩

更高级的策略是对历史消息进行语义压缩，保留关键信息的同时减少 token 消耗。如您代码中实现的 compression_messages 函数：

import json
import os
import uuid
from datetime import datetime
from app.core.logging import logger
import redis
from fastapi import APIRouter
from openai import OpenAI

from app.schemas.llm_case1 import LLMCase2

llm_day02_router = APIRouter(
prefix="/llm-day02",
tags=["llm_day02"],
)

redis_client = redis.Redis(
host="localhost",
port=6379,
db=3,
decode_responses=True
)

@llm_day02_router.get("/create_session",summary="创建会话")
async def create_session(user_id:int):
key = f"bossfastApiProject:llm:session:{user_id}"

current_session = {
"session_id":str(uuid.uuid4()),
"title":"新对话",
"create_time":datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}

redis_client.lpush(key,json.dumps(current_session,ensure_ascii=False))
return {"code":1,"message":"创建会话成功","data": {"current_session":current_session}}

@llm_day02_router.get("/get_session_list",summary="获取会话列表")
async def get_session_list(user_id:int):
key = f"bossfastApiProject:llm:session:{user_id}"
user_sessions = [json.loads(session) for session in redis_client.lrange(key,0,-1)]
return {"code":1,"message":"获取会话列表成功","data": {"user_sessions":user_sessions}}

client = OpenAI(
# 建议通过环境变量配置API Key，避免硬编码。
api_key=os.environ["DASHSCOPE_API_KEY"],
# API Key与地域强绑定，请确保base_url与API Key的地域一致。
base_url="https://llm-cn7w8jk1i8zqovoz.cn-beijing.maas.aliyuncs.com/compatible-mode/v1"
)

def get_ai_response(messages: list):
completion = client.chat.completions.create(
model="qwen3.7-plus",
messages=messages
)

return completion.choices[0].message.content

def compression_message(messages: list):
logger.info(f"当前的消息长度为：{len(messages)}")
if len(messages) <= 2:
logger.info("消息数量小于2，无需压缩")
return messages
logger.info("开始压缩消息")
recently_messages = messages[-2:]
history_messages = messages[:-2]
c_messages = [{
"role": "user","content": f"请将以下用户与大模型的上下文进行语义压缩，要保留关键信息，上下文：{history_messages}"
}]
logger.info(f"压缩后的消息：{c_messages}")
res = get_ai_response(c_messages)
return [{"role": "user", "content": res}] + recently_messages

@llm_day02_router.post("/send_message",summary="发送消息")
async def send_message(case_request:LLMCase2):
key = f"bossfastApiProject:llm:case2:{case_request.user_id}:{case_request.session_id}"
user_session_messages = redis_client.lrange(key,0,-1)
print(f"当前会话的消息{user_session_messages}")
print(f"当前会话的消息长度{len(user_session_messages)}")
if len(user_session_messages) == 0:
print("这是第一轮对话")
system_message = {
"role": "system",
"content": "你是一个智能助手"
}
redis_client.rpush(key,json.dumps(system_message,ensure_ascii=False))
session_key = f"bossfastApiProject:llm:session:{case_request.user_id}"
user_sessions = redis_client.lrange(session_key,0,-1)
for index,session in enumerate(user_sessions):
session_dict = json.loads(session)
if session_dict["session_id"] == case_request.session_id:
session_dict["title"] = case_request.message[:10] + "…"
redis_client.lset(session_key,index,json.dumps(session_dict,ensure_ascii=False))
current_user_message = {
"role": "user",
"content": case_request.message,
"time":datetime.now().strftime("%Y-%m-%d %H:%M:%S")

}
redis_client.rpush(key,json.dumps(current_user_message,ensure_ascii=False))
messages = [json.loads(message) for message in redis_client.lrange(key,0,-1)]
print(messages)

messages = compression_message(messages)
logger.info(f"接受压缩消息{messages}")
api_reply = get_ai_response(messages)
api_reply_message = {
"role": "assistant",
"content": api_reply,
"time":datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
redis_client.rpush(key,json.dumps(api_reply_message,ensure_ascii=False))
return {"code":1,"message":"发送成功","data": {"api_reply":api_reply}}

@llm_day02_router.get("/get_user_session_message",summary="查询会话详情")
async def get_user_session_message(user_id:int,session_id:str):
key = f"bossfastApiProject:llm:case2:{user_id}:{session_id}"
user_session_messages = redis_client.lrange(key,1,-1)
user_session_messages = [json.loads(message) for message in user_session_messages]
return {"code":1,"message":"查询会话详情成功","data": {"user_session_messages":user_session_messages}}

四、总结

本文详细介绍了阿里云通义千问 API 在多轮对话中的实现原理和最佳实践：

• 无状态设计：通义千问 API 是无状态的，开发者需要在每次请求中显式传递完整的对话上下文。
• 消息结构：通过 messages 数组维护对话历史，包含用户、助手和系统角色的消息。
• 上下文管理：随着对话轮次增加，需要采用滑动窗口、历史消息压缩等策略来管理上下文长度。
• 实践建议：根据实际场景选择合适的上下文管理策略，平衡对话连贯性与性能开销。

通过合理设计消息传递机制和上下文管理策略，可以构建出高效、稳定的多轮对话应用，充分发挥大语言模型的对话能力。