主题
原文：https://www.mstx.cn/harness/429.html · 创建：2026-09-06 · 更新：2026-09-06 · 阅读：1090
一个容易被忽视的问题
看下面这段对话：
用户：帮我查一下明天的天气
Agent：明天北京晴，26°C。
用户：那后天呢？
Agent：抱歉，我不确定您想查询哪个城市的天气。请告诉我城市名称。Agent的回答很糟糕。用户显然是在延续上一个话题——查询北京的天气，只是日期从"明天"变成了"后天"。但Agent"忘记"了对话的上下文，把"那后天呢"当作一个独立的新问题来处理。
这暴露了一个关键问题：多轮对话中的信息是渐进的、依赖的、省略的。 用户不会在每一轮都重复完整的信息——他们假设Agent记得之前说过什么。当Agent"忘记"时，对话的流畅性就被打破了。
会话状态管理的目标就是：让Agent在跨多轮对话时保持对上下文的"意识"——知道话题是什么、已经确认了什么、还需要补充什么。
会话状态的多层结构
一个完整的会话状态包含三个层次：
| 层次 | 包含字段 |
|---|---|
| 会话层（Session Level） | · 会话ID · 开始时间 · 用户身份 · 当前话题 |
| 任务层（Task Level） | · 当前任务目标 · 已完成的子任务 · 待确认的信息 · 任务的中间结果 |
| 消息层（Message Level） | · 对话历史（短期记忆） · 每条消息的角色和内容 |
消息层在5.1节中已经处理过了（短期记忆），会话层和任务层是本节的重点。它们是消息之上的结构化抽象——不直接存储消息内容，而是跟踪对话的"元信息"。
实现会话状态管理
第一步：定义会话和任务状态
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
from enum import Enum
import uuid
class TaskStatus(Enum):
"""任务状态"""
NOT_STARTED = "not_started"
IN_PROGRESS = "in_progress"
WAITING_CONFIRMATION = "waiting_confirmation"
WAITING_INPUT = "waiting_input"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class ConversationTopic(Enum):
"""对话话题类型"""
GENERAL_QUERY = "general_query"
TASK_EXECUTION = "task_execution"
CLARIFICATION = "clarification"
FOLLOW_UP = "follow_up"
ERROR_RECOVERY = "error_recovery"
@dataclass
class TaskState:
"""一个任务的执行状态"""
task_id: str
description: str
status: TaskStatus
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
updated_at: str = field(default_factory=lambda: datetime.now().isoformat())
subtasks_completed: List[str] = field(default_factory=list)
pending_confirmations: List[str] = field(default_factory=list)
pending_inputs: List[str] = field(default_factory=list)
result: Optional[str] = None
error: Optional[str] = None
@dataclass
class SessionState:
"""一次会话的完整状态"""
session_id: str
user_id: Optional[str] = None
started_at: str = field(default_factory=lambda: datetime.now().isoformat())
last_activity_at: str = field(default_factory=lambda: datetime.now().isoformat())
current_topic: ConversationTopic = ConversationTopic.GENERAL_QUERY
current_task: Optional[TaskState] = None
task_history: List[TaskState] = field(default_factory=list)
confirmed_facts: Dict[str, Any] = field(default_factory=dict)
unresolved_questions: List[str] = field(default_factory=list)
conversation_turns: int = 0
metadata: Dict[str, Any] = field(default_factory=dict)
def touch(self):
"""更新最后活动时间"""
self.last_activity_at = datetime.now().isoformat()
self.conversation_turns += 1第二步：实现状态管理器
python
class SessionStateManager:
"""会话状态管理器"""
def __init__(self, session_timeout_minutes: int = 30):
self.sessions: Dict[str, SessionState] = {}
self.session_timeout_minutes = session_timeout_minutes
def create_session(self, user_id: Optional[str] = None) -> SessionState:
"""创建一个新会话"""
session = SessionState(
session_id=str(uuid.uuid4())[:12],
user_id=user_id,
)
self.sessions[session.session_id] = session
return session
def get_session(self, session_id: str) -> Optional[SessionState]:
"""获取会话状态"""
session = self.sessions.get(session_id)
if session is None:
return None
# 检查会话是否超时
last_activity = datetime.fromisoformat(session.last_activity_at)
elapsed = (datetime.now() - last_activity).total_seconds() / 60
if elapsed > self.session_timeout_minutes:
self.close_session(session_id)
return None
return session
def close_session(self, session_id: str):
"""关闭会话"""
session = self.sessions.pop(session_id, None)
if session:
# 将会话归档（实际场景中会持久化到数据库）
self._archive_session(session)
def _archive_session(self, session: SessionState):
"""归档会话（此处简化为打印统计）"""
print(f"会话 {session.session_id} 已归档。"
f"共{session.conversation_turns}轮对话，"
f"完成{len(session.task_history)}个任务。")
def start_task(self, session: SessionState, description: str) -> TaskState:
"""开始一个新任务"""
task = TaskState(
task_id=str(uuid.uuid4())[:8],
description=description,
status=TaskStatus.IN_PROGRESS,
)
session.current_task = task
session.current_topic = ConversationTopic.TASK_EXECUTION
session.touch()
return task
def update_task(self, session: SessionState,
status: Optional[TaskStatus] = None,
result: Optional[str] = None,
error: Optional[str] = None):
"""更新当前任务的状态"""
if session.current_task is None:
return
task = session.current_task
if status:
task.status = status
if result:
task.result = result
if error:
task.error = error
task.updated_at = datetime.now().isoformat()
session.touch()
# 如果任务结束，移入历史
if task.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED):
session.task_history.append(task)
session.current_task = None
session.current_topic = ConversationTopic.GENERAL_QUERY
def add_confirmed_fact(self, session: SessionState, key: str, value: Any):
"""添加一个已确认的事实"""
session.confirmed_facts[key] = value
session.touch()
def add_pending_confirmation(self, session: SessionState, item: str):
"""添加一个待确认项"""
if session.current_task:
session.current_task.pending_confirmations.append(item)
session.touch()
def resolve_confirmation(self, session: SessionState, item: str,
confirmed: bool, value: Any = None):
"""解决一个待确认项"""
if session.current_task and item in session.current_task.pending_confirmations:
session.current_task.pending_confirmations.remove(item)
if confirmed and value is not None:
session.add_confirmed_fact(session, item, value)
session.touch()
def get_context_summary(self, session: SessionState) -> str:
"""生成会话状态的文本摘要（用于注入LLM上下文）"""
lines = []
# 当前话题
lines.append(f"当前话题：{session.current_topic.value}")
# 当前任务
if session.current_task:
task = session.current_task
lines.append(f"当前任务：{task.description}（状态：{task.status.value}）")
if task.subtasks_completed:
lines.append(f"已完成子任务：{', '.join(task.subtasks_completed)}")
if task.pending_confirmations:
lines.append(f"待确认：{', '.join(task.pending_confirmations)}")
if task.pending_inputs:
lines.append(f"待输入：{', '.join(task.pending_inputs)}")
# 已确认的事实
if session.confirmed_facts:
facts = [f"{k}={v}" for k, v in session.confirmed_facts.items()]
lines.append(f"已确认信息：{'；'.join(facts)}")
# 未解决的问题
if session.unresolved_questions:
lines.append(f"待解决的问题：{', '.join(session.unresolved_questions)}")
return "\n".join(lines) if lines else "（无特殊上下文）"第三步：集成到MemorySystem中
修改MemorySystem，加入会话状态管理的支持：
python
class MemorySystem:
"""完整的记忆系统：短期 + 长期 + 会话状态"""
def __init__(self, system_prompt: str,
long_term_memory: Optional[LongTermMemory] = None,
state_manager: Optional[SessionStateManager] = None):
self.system_prompt = system_prompt
self.state_manager = state_manager or SessionStateManager()
self.long_term = long_term_memory
self.extractor = MemoryExtractor()
self.session: Optional[SessionState] = None
self.short_term: Optional[WorkingMemory] = None
self.messages: List[Dict[str, Any]] = []
def initialize(self, user_input: str):
"""初始化一次新的会话"""
# 创建或恢复会话
self.session = self.state_manager.create_session()
# 初始化短期记忆
self.short_term = WorkingMemory(max_tokens=8000)
self.short_term.set_system_prompt(self.system_prompt)
# 添加用户输入
self.short_term.add_user_message(user_input)
# 注入长期记忆（如5.2节所述）
if self.long_term:
relevant = self.long_term.retrieve(user_input, top_k=3)
if relevant:
context_text = self.long_term.retrieve_as_context(user_input, top_k=3)
self.short_term.add_user_message(
f"[系统注入了相关的历史记忆]\n{context_text}"
)
# 注入会话状态摘要
state_summary = self.state_manager.get_context_summary(self.session)
if state_summary != "（无特殊上下文）":
self.short_term.add_user_message(
f"[当前会话状态]\n{state_summary}"
)
self.messages = self.short_term.get_context()
def continue_conversation(self, user_input: str):
"""继续当前会话（多轮对话的后续轮次）"""
if self.session is None:
self.initialize(user_input)
return
# 分析用户输入，更新会话状态
self._update_state_from_user_input(user_input)
# 添加用户消息到短期记忆
self.short_term.add_user_message(user_input)
# 注入更新后的会话状态摘要
state_summary = self.state_manager.get_context_summary(self.session)
if state_summary != "（无特殊上下文）":
self.short_term.add_user_message(
f"[更新后的会话状态]\n{state_summary}"
)
self.messages = self.short_term.get_context()
def _update_state_from_user_input(self, user_input: str):
"""从用户输入中分析并更新会话状态"""
if self.session is None:
return
# 检测后续跟进类输入
followup_patterns = [
"那", "然后", "还有", "另外", "顺便",
"如果", "能不能", "再", "也",
]
is_followup = any(user_input.startswith(p) for p in followup_patterns)
if is_followup and self.session.current_task:
# 这是对当前任务的跟进
self.session.current_topic = ConversationTopic.FOLLOW_UP
elif is_followup and self.session.task_history:
# 可能涉及之前完成的任务
self.session.current_topic = ConversationTopic.FOLLOW_UP
elif "?" in user_input or "吗" in user_input or "什么" in user_input:
# 可能是澄清问题
self.session.current_topic = ConversationTopic.GENERAL_QUERY
else:
# 可能是一个新任务
self.session.current_topic = ConversationTopic.TASK_EXECUTION
self.session.touch()
def add_tool_result(self, result):
"""添加工具结果，并更新会话状态"""
self.short_term.add_tool_result(
getattr(result, 'tool_call_id', ""),
result.content
)
# 从工具结果中提取确认的事实
if self.session and getattr(result, 'success', False):
tool_name = getattr(result, 'tool_name', '')
if tool_name:
self.state_manager.add_confirmed_fact(
self.session,
f"last_{tool_name}_result",
result.content[:100],
)
self.messages = self.short_term.get_context()
def get_context(self) -> List[Dict[str, Any]]:
"""获取当前上下文"""
return self.short_term.get_context() if self.short_term else []处理对话中的指代消解
多轮对话中最大的挑战之一是指代——用户说"它"、"那个"、"这个"时，指的是什么？
用户：帮我查一下ORD12345678的物流状态
Agent：订单已发货，预计周五送达。
用户：那它现在到哪了？ ← "它"指的是"ORD12345678"处理指代消解的策略：在会话状态中维护"当前讨论对象"。
python
class SessionStateManager:
def __init__(self, session_timeout_minutes: int = 30):
# ... 之前的初始化 ...
self.current_subject: Dict[str, Any] = {}
def set_current_subject(self, session: SessionState,
subject_type: str, subject_id: str,
description: str):
"""设置当前讨论的对象"""
session.metadata["current_subject"] = {
"type": subject_type,
"id": subject_id,
"description": description,
}
session.touch()
def get_current_subject(self, session: SessionState) -> Optional[Dict[str, Any]]:
"""获取当前讨论的对象"""
return session.metadata.get("current_subject")
def resolve_reference(self, session: SessionState,
user_input: str) -> Optional[Dict[str, Any]]:
"""尝试解析用户输入中的指代"""
# 检测指代词
reference_words = ["它", "那个", "这个", "那", "这"]
has_reference = any(word in user_input for word in reference_words)
if has_reference:
return self.get_current_subject(session)
return None在MemorySystem中集成指代消解：
python
def _update_state_from_user_input(self, user_input: str):
"""更新会话状态，包含指代消解"""
if self.session is None:
return
# 尝试解析指代
reference = self.state_manager.resolve_reference(self.session, user_input)
if reference:
# 用户使用了指代，将当前讨论对象注入到输入中
augmented_input = (
f"[指代解析：用户说的'{user_input}'中的指代词可能指的是："
f"{reference['description']}（{reference['type']}:{reference['id']}）]\n"
f"原始输入：{user_input}"
)
self.short_term.add_user_message(augmented_input)
else:
self.short_term.add_user_message(user_input)
# ... 其他状态更新逻辑 ...处理话题切换
多轮对话中，用户可能突然切换话题：
用户：帮我查一下明天的天气
Agent：明天北京晴。
用户：对了，帮我创建个任务，周五前交周报Agent需要识别话题的切换，并更新会话状态。如果Agent还在"天气"话题的上下文中理解"任务创建"，可能会产生困惑。
python
class TopicDetector:
"""话题切换检测器"""
def __init__(self):
self.topic_keywords = {
"weather": ["天气", "下雨", "温度", "晴", "阴"],
"task": ["任务", "创建", "待办", "完成"],
"calendar": ["日程", "会议", "安排", "日历"],
"message": ["发送", "通知", "消息", "告诉"],
"search": ["搜索", "查找", "文档", "资料"],
}
def detect_topic(self, user_input: str) -> str:
"""检测用户输入的话题"""
scores = {}
for topic, keywords in self.topic_keywords.items():
score = sum(1 for kw in keywords if kw in user_input)
if score > 0:
scores[topic] = score
if not scores:
return "general"
return max(scores, key=scores.get)
def is_topic_switch(self, previous_topic: str, current_topic: str) -> bool:
"""判断是否发生了话题切换"""
if previous_topic == "general" or current_topic == "general":
return False
return previous_topic != current_topic多轮对话的完整流程示例 以下是一个完整的对话流程，展示会话状态如何帮助Agent维持连续性：
【第1轮】
用户：帮我查一下ORD12345678的物流状态
状态更新：创建任务"查询订单ORD12345678的物流状态"（IN_PROGRESS）
Agent调用：check_order_status(order_id="ORD12345678")
工具返回：订单已发货，当前在杭州中转站，预计明天送达北京
状态更新：任务完成（COMPLETED），设置current_subject={type:"order", id:"ORD12345678"}
Agent回答：您的订单已发货，目前在杭州中转站，预计明天送达北京。
【第2轮】
用户：那它到北京之后什么时候派送？
指代解析："它"→当前讨论对象ORD12345678
状态更新：话题=FOLLOW_UP，当前任务=查询派送时间
Agent调用：get_delivery_estimate(order_id="ORD12345678")
工具返回：预计明天上午派送
Agent回答：预计明天上午开始派送。
【第3轮】
用户：明天上午我要开会，帮我查一下明天上午的日程
话题检测：从"order"切换到"calendar"
状态更新：话题切换，创建新任务
Agent调用：query_calendar(start_date="tomorrow", end_date="tomorrow")
工具返回：明天上午10:00-11:30有产品评审会
Agent回答：明天上午10:00-11:30有产品评审会。如果您需要调整派送时间，可以联系快递员。注意Agent在第3轮的表现：它没有忘记之前关于订单的讨论，在回答日程后还提到了派送时间的关联。这种"全局意识"正是会话状态管理的价值。
会话状态的持久化
会话状态不应只存在内存中。如果Agent服务重启，所有会话状态都丢失，用户会突然发现Agent"失忆"了。
python
import json
import os
class PersistentSessionStore:
"""持久化的会话存储"""
def __init__(self, storage_dir: str = "./session_storage"):
self.storage_dir = storage_dir
os.makedirs(storage_dir, exist_ok=True)
def save(self, session: SessionState):
"""保存会话到磁盘"""
filepath = os.path.join(self.storage_dir, f"{session.session_id}.json")
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(session_to_dict(session), f, ensure_ascii=False, indent=2)
def load(self, session_id: str) -> Optional[SessionState]:
"""从磁盘加载会话"""
filepath = os.path.join(self.storage_dir, f"{session_id}.json")
if not os.path.exists(filepath):
return None
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
return session_from_dict(data)
def delete(self, session_id: str):
"""删除会话"""
filepath = os.path.join(self.storage_dir, f"{session_id}.json")
if os.path.exists(filepath):
os.remove(filepath)
def session_to_dict(session: SessionState) -> Dict[str, Any]:
"""将会话状态序列化为字典"""
# 简化的序列化
return {
"session_id": session.session_id,
"user_id": session.user_id,
"started_at": session.started_at,
"last_activity_at": session.last_activity_at,
"current_topic": session.current_topic.value,
"confirmed_facts": session.confirmed_facts,
"conversation_turns": session.conversation_turns,
"metadata": session.metadata,
}本节小结
会话状态管理让Agent在多轮对话中保持连续性。本节讨论了：
- 会话状态的三个层次：会话层、任务层、消息层
- 任务跟踪：当前任务、已完成子任务、待确认项
- 指代消解：将"它"、"那个"解析为具体的对象
- 话题切换检测：识别用户是否转移了话题
- 状态持久化：跨重启保持会话状态
核心原则：会话状态管理的本质是让Agent知道"我们在聊什么、聊到哪了、还需要什么"。它不是替代对话历史，而是为对话历史提供结构化的元信息，帮助LLM更准确地理解上下文。
在下一节中，我们将讨论记忆压缩与摘要策略——当上下文窗口面临压力时，如何智能地压缩历史信息，保持Agent的推理质量。