LV005-有状态的对话

实验2：有状态对话的快照验证

一、实验详情

实验仓库：learning-ai/ai-course-lab-basic

目标：实现支持多会话隔离的对话系统

文件位置：student_code/lab2/main.py

需要实现：chat_with_memory() 函数

任务要求：

• 实现多会话隔离的对话系统
• 正确计算 history_length（本次对话前的消息数）
• 将历史消息作为上下文传递给模型
• 保存用户消息和 AI 回复到历史记录

关键要求：

• 返回字典，包含 response, history_length, session_id
• 不同 session_id 的历史完全独立
• history_length 必须精确反映本次对话前的消息数
• 历史消息作为上下文传递给模型

实现提示：

# 当前状态：raise NotImplementedError("请实现 chat_with_memory 函数")
# 你需要：
# 1. 初始化 SESSION_HISTORY [session_id] = []
# 2. history_length = len(history) # 在保存新消息前计算
# 3. 构建包含历史的 prompt
# 4. 调用 Ollama API 并保存结果到 SESSION_HISTORY

数据结构建议：

SESSION_HISTORY = {
    "session_001": [
        {"role": "user", "content": "你好"},
        {"role": "assistant", "content": "你好！..."},
    ]
}

测试运行：

pytest grader/test_lab2.py -v

测试用例：

• ✅ 首次对话 history_length = 0（15%）
• ✅ 历史消息累积（20%）
• ✅ 会话隔离验证（25%）
• ✅ 交叉会话测试（25%）
• ✅ 输出结构验证（15%）
• 🌟 内容持久化验证（加分项）

二、环境准备

1. 系统要求

• Python: 3.10 或更高版本
• 操作系统: Linux / macOS / Windows
• Ollama: 本地 AI 模型服务

2. 安装 Ollama

2.1 Linux / macOS

# 安装 Ollama
curl -fsSL https://ollama.ai/install.sh | sh

# 启动 Ollama 服务
ollama serve

# 下载 Qwen3-8B 模型（新开一个终端）
ollama pull qwen3:8b

2.2 Windows

1. 访问 Ollama 官网 下载 Windows 安装包
2. 安装并启动 Ollama
3. 在命令提示符中运行：ollama pull qwen3:8b

3. 验证 Ollama 服务

# 检查 Ollama 是否运行
curl http://localhost:11434/api/tags

# 应该返回包含 qwen3:8b 的模型列表

4. 安装 Python 依赖

# 安装依赖
pip install -r requirements.txt

requirements.txt 内容如下：

pydantic>=2.0.0
pytest>=7.4.0
pytest-timeout>=2.1.0
pytest-json-report>=1.5.0
httpx>=0.25.0

三、实验示例

实现的关键就是定义一个全局变量，把每一次用户输入的内容和大模型回复的的内容存到一起，作为下一次发送给大模型的数据内容。对于多个会话，就是定义多个这种全局变量来分别记录每个会话的内容。

1. main.py

"""
实验2：有状态对话的快照验证
学生需要实现 chat_with_memory 函数, 管理多个独立会话的历史记录
"""
from typing import Dict, List
import httpx

# 全局会话存储：{session_id: [消息列表]}
# 每条消息格式: {"role": "user" | "assistant", "content": "消息内容"}
SESSION_HISTORY: Dict[str, List[Dict[str, str]]] = {}


def chat_with_memory(message: str, session_id: str) -> dict:
    """
    带记忆功能的对话函数, 支持多会话隔离
    
    参数:
        message: 用户当前消息
        session_id: 会话唯一标识符
    
    返回:
        字典, 包含以下键:
        - response (str): AI的回复内容
        - history_length (int): 当前会话在本次对话前的历史消息数量
        - session_id (str): 回显的会话ID
    
    实现要求:
        1. 不同 session_id 的对话历史必须完全独立
        2. 同一 session_id 的多次调用必须累积历史
        3. history_length 必须精确反映本次对话前的历史消息数（user+assistant成对计数）
        4. 调用模型前, 将历史消息作为上下文拼接到 Prompt 中
    
    数据结构建议:
        SESSION_HISTORY = {
            "session_001": [
                {"role": "user", "content": "你好"},
                {"role": "assistant", "content": "你好！有什么可以帮助你的吗？"},
                {"role": "user", "content": "天气怎么样"},
                {"role": "assistant", "content": "抱歉..."}
            ]
        }
    """
    global SESSION_HISTORY

    # 1. 检查 session_id 是否存在, 不存在则初始化空列表
    if session_id not in SESSION_HISTORY:
        SESSION_HISTORY[session_id] = []
    # 2. 计算 history_length（当前会话在本次对话前的消息数）
    history_length = len(SESSION_HISTORY[session_id])
    # 3. 构建 Prompt, 包含历史上下文 + 当前消息
    context = SESSION_HISTORY[session_id]
    prompt_parts = []
    # 3.1 添加历史消息(包括用户之前发送的和大模型回复的)
    for tmp in context:
        if tmp["role"] == "user":
            prompt_parts.append(f"user: {tmp['content']}")
        else:
            prompt_parts.append(f"assistant: {tmp['content']}")
    # 3.2 添加当前消息(用户发送的消息)
    prompt_parts.append(f"user: {message}")
    # 3.3 构建完整 Prompt
    full_prompt = "\n".join(prompt_parts)
    # 4. 调用 Ollama API 获取回复
    try:
        response = httpx.post(
            "http://localhost:11434/api/generate",
            json = {
                "model": "qwen3:0.6b",
                "prompt": full_prompt,
                "stream": False
            },
            timeout = 30
        )
        response.raise_for_status()
    except httpx.RequestError as e:
        raise Exception(f"Unable to connect to the Ollama service: {e}")
    except httpx.HTTPStatusError as e:
        raise Exception(f"API request failed: {e}")
    # 4.1 提取回复内容
    result = response.json()
    assistant_response = result.get("response", "")
    # 5. 保存用户消息和助手回复到历史记录
    SESSION_HISTORY[session_id].append({
        "role": "user",
        "content": message
    })
    SESSION_HISTORY[session_id].append({
        "role": "assistant",
        "content": assistant_response
    })
    # 6. 返回结构化结果
    return {
        "response": assistant_response,
        "history_length": history_length,
        "session_id": session_id
    }

def clear_session(session_id: str = None):
    """
    清除会话历史（辅助函数, 用于测试）
    
    参数:
        session_id: 要清除的会话ID, 如果为 None 则清除所有会话
    """
    global SESSION_HISTORY
    
    if session_id is None:
        SESSION_HISTORY.clear()
    elif session_id in SESSION_HISTORY:
        del SESSION_HISTORY[session_id]


# 测试代码（可选, 用于学生本地调试）
if __name__ == "__main__":
    # 清空历史
    clear_session()
    
    # 测试单会话
    print("=== 测试单会话 ===")
    session_id = "test_session"
    
    result1 = chat_with_memory("你好", session_id)
    print(f"第1次对话 - history_length: {result1['history_length']}, response: {result1['response'][:50]}")
    
    result2 = chat_with_memory("我叫张三", session_id)
    print(f"第2次对话 - history_length: {result2['history_length']}, response: {result2['response'][:50]}")
    
    result3 = chat_with_memory("我刚才说了什么？", session_id)
    print(f"第3次对话 - history_length: {result3['history_length']}, response: {result3['response'][:50]}")
    
    # 测试多会话隔离
    print("\n=== 测试会话隔离 ===")
    clear_session()
    
    result_a1 = chat_with_memory("苹果", "session_A")
    print(f"Session A 第1次 - history_length: {result_a1['history_length']}")
    
    result_b1 = chat_with_memory("香蕉", "session_B")
    print(f"Session B 第1次 - history_length: {result_b1['history_length']}")
    
    result_a2 = chat_with_memory("我之前说了什么？", "session_A")
    print(f"Session A 第2次 - history_length: {result_a2['history_length']}, response: {result_a2['response'][:50]}")

2. 运行结果

python .\main.py

【例】