4. 支持多轮对话状态管理

任务目标

在AI Agent Harness中实现一个带会话状态管理的多轮对话能力：同一会话ID的多次请求能共享上下文，不同会话ID之间相互隔离。核心是让Agent在后续轮次中能引用前文信息，而不是每次从零开始。

核心实现

采用内存态会话存储 + 消息历史注入的最小闭环方案。Harness维护一个SessionStore，以会话ID为键保存消息列表；每次请求到达时，将历史消息与当前输入合并后交给Agent处理，再将本轮交互追加回存储。为便于验证，Agent采用基于关键词的简单应答逻辑，不依赖外部LLM服务。

# harness_session.py
from dataclasses import dataclass, field
from typing import Dict, List
import uuid

@dataclass
class Session:
    session_id: str
    messages: List[Dict] = field(default_factory=list)

class SessionStore:
    """内存会话存储，生产环境可替换为Redis等持久化实现"""
    def __init__(self):
        self._sessions: Dict[str, Session] = {}

    def get_or_create(self, session_id: str) -> Session:
        if session_id not in self._sessions:
            self._sessions[session_id] = Session(session_id=session_id)
        return self._sessions[session_id]

    def add_message(self, session_id: str, role: str, content: str):
        session = self.get_or_create(session_id)
        session.messages.append({"role": role, "content": content})

    def get_history(self, session_id: str) -> List[Dict]:
        return self.get_or_create(session_id).messages

class AgentHarness:
    """支持多轮会话状态的Agent执行环境"""
    def __init__(self):
        self.store = SessionStore()

    def _agent_reply(self, user_input: str, history: List[Dict]) -> str:
        # 演示用规则引擎：能识别历史中的名字与话题
        if "我叫" in user_input:
            return f"你好 {user_input.split('我叫')[1].strip()}！我已记住你。"
        if "我住在" in user_input:
            city = user_input.split("我住在")[1].strip()
            return f"好的，你住在{city}。"
        # 检查历史中是否有名字信息
        for msg in history:
            if msg["role"] == "user" and "我叫" in msg["content"]:
                name = msg["content"].split("我叫")[1].strip()
                if "住" in user_input:
                    return f"{name}，你刚才提到居住信息了吗？"
                return f"{name}，我在听，请继续说。"
        return "请告诉我你的名字或你住在哪里，我才能记住。"

    def chat(self, session_id: str, user_input: str) -> str:
        history = self.store.get_history(session_id)
        reply = self._agent_reply(user_input, history)
        # 更新会话状态
        self.store.add_message(session_id, "user", user_input)
        self.store.add_message(session_id, "assistant", reply)
        return reply

# 最小调用入口
if __name__ == "__main__":
    harness = AgentHarness()
    sid = uuid.uuid4().hex  # 模拟前端生成的会话ID

    # 第一轮：自我介绍
    r1 = harness.chat(sid, "我叫李明")
    # 第二轮：跨轮引用名字
    r2 = harness.chat(sid, "我住在上海")
    # 第三轮：验证记忆
    r3 = harness.chat(sid, "你还记得我吗？")

    print("R1:", r1)
    print("R2:", r2)
    print("R3:", r3)

    # 验证会话隔离
    sid2 = uuid.uuid4().hex
    r_other = harness.chat(sid2, "你还记得我吗？")
    print("Other session:", r_other)

运行与验证

python harness_session.py

预期输出（确定性逻辑推导）：

R1: 你好 李明！我已记住你。
R2: 好的，你住在上海。
R3: 李明，我在听，请继续说。
Other session: 请告诉我你的名字或你住在哪里，我才能记住。

通过判定标准：

• R3能输出“李明”说明第一轮信息被保留并注入到第三轮上下文；
• 新会话sid2的回答不包含“李明”，证明会话间状态隔离；
• 若将SessionStore替换为Redis实现，需保证get_or_create与add_message接口不变，即可无缝迁移。

关键说明

本实现将SessionStore作为Harness的独立状态层，Agent本身无状态，通过每次请求注入完整历史实现记忆。实际接入LLM时，只需将_agent_reply替换为对模型API的调用，并把history作为messages参数传入即可。会话过期策略、持久化与并发锁属于生产环境增强项，不在本最小闭环内展开。