语音交互场景下“说完就忘”简直是语音 Agent 落地过程中最让人头疼的问题。用户明明在一分钟前说过“帮我订靠窗的位置”等 Agent 确认时间时却表现得像第一次听到一样。本文围绕 voice agent 的上下文遗忘问题完整拆解一种“内心独白inner monologue”机制它是如何工作的、如何设计状态结构、怎么和 ASR / LLM / TTS 串联以及落地时有哪些坑。全程包含结构设计、可运行示例代码、常见问题排查和工程建议适合正在做语音助手、Agent 应用或对话系统的开发者。1. 背景语音 Agent 为什么会“忘事”1.1 多轮语音对话的上下文困境先看一个很自然的用户交互用户帮我查一下明天上午 10 点到 12 点的空闲会议室。 Agent好的我找到了 3 间A302 靠窗B101 有投影C205 空着。 用户那就订 A302 吧。 Agent好的需要我同时帮你预定投影设备吗 用户等下刚才那间靠窗的会议室是明天几点来着如果 Agent 在这一轮“失忆”了用户体验会非常割裂。语音 Agent 和文字聊天机器人不同它有以下几个天然难点会话连续性弱用户可能中途打断、换话题、很久之后又切回前面的任务模型很难知道当前该以哪条对话主线为准。上下文窗口有限即便模型支持很长的上下文把每一轮语音转写文本全部塞进 Prompt也会导致成本升高、响应变慢甚至把关键信息淹没在噪音里。语音转写有噪声ASR 可能出错一句“靠窗”可能被识别成“靠场”如果 Agent 只依赖原始转写文本很容易被带偏。交互节奏快语音场景讲究低延迟不能像文字对话那样让用户等很久。说到底语音 Agent 缺的不是“模型能力”而是一个自己能随时查看、能持续更新的“工作记忆”。这正是本文要讲的 inner monologue 机制可以解决的问题。1.2 Inner Monologue 是什么Inner Monologue直译是“内心独白”。在普通对话里一个人在回答别人之前通常会在心里快速整理一下对方刚才想干什么 我已经确认了哪些信息 还差什么信息 我现在应该问什么语音 Agent 的 inner monologue 就是把这个“心里整理”的过程显式建模出来。它是在用户输入和 Agent 输出之间插入的一段结构化内部状态记录当前任务的目标、已收集的信息、待确认的细节、历史关键决策等。与直接拼接“完整聊天历史”不同inner monologue 是一种“压缩后”的记忆。它不保存每一帧语音文本而是保存这个 Agent 在当前会话中真正需要记住的事实。举个例子用户订会议室这个场景下Agent 的内心独白可能长这样{ current_task: 预订会议室 A302, known_info: { date: 明天, time_range: 10:00-12:00, room: A302 }, pending_questions: [是否需要投影设备], session_goal: 完成会议室预订, user_preference: { window_seat: true } }当用户问“刚才那间靠窗的会议室是明天几点来着”时Agent 不需要翻历史记录直接查 inner monologue 里的known_info就能回答“A302明天上午 10 点到 12 点”。1.3 为什么不是“把所有历史都拼进 Prompt”有的开发者会问“直接把前 N 轮对话全部拼进 Prompt 不就行了吗”这种做法在短时间内有用但长期来看问题不少成本爆炸每多一轮Token 消耗就增加一部分语音场景高频调用费用增长很快。延迟升高Prompt 越长模型生成首字时间越晚用户会觉得“AI 反应变慢”。信息稀释关键信息淹没在一堆“嗯”“然后”“那个”等口语词里模型反而提取不准。超出窗口对话一旦变长最终会顶到上下文上限必须做裁剪而简单裁剪会丢失重要信息。Inner monologue 的核心思路是“先提炼再记忆”。它把“记住什么”这件事交给一个显式的状态管理模块去完成而不是把所有原始数据一股脑丢给模型。这样模型每次只需要读一份紧凑的内部状态外加当前用户输入就能稳定地产出高质量回复。2. 整体架构设计2.1 一条语音请求的处理链路一个带 inner monologue 的 voice agent其处理链路大致如下用户说话 - ASR语音转文字 - Agent Core更新 inner monologue决定回复内容 - LLM根据内心独白生成回复文本 - TTS文本转语音 - 用户听到回复这里的核心是Agent Core它负责两部分工作在每轮对话开始前把当前的 inner monologue 和本次用户输入一起交给 LLM。在每轮对话结束后根据 LLM 的输出和用户输入更新 inner monologue。如果把Agent Core拆细流程可以写成接收 ASR 输出的文本。读取当前会话的 inner state。把 inner state 当前输入 任务提示词一起发送给 LLM。从 LLM 返回结果中解析两个部分reply对用户说的话和state_update对内心独白的更新指令。执行状态更新写入新的 inner state。把reply送入 TTS。2.2 Inner Monologue 模块的职责Inner Monologue 模块并不是一个独立的大模型服务它更像一个“状态机 记忆数据库”。它负责维护 Session 级状态每个用户会话对应一份独立的 inner state。解析意图与信息槽位从用户输入中提取关键实体。维护待确认问题列表哪些信息还没拿到。记录已完成步骤任务执行到哪一步。处理状态回滚用户反悔、修改需求时能回到之前的状态。2.3 状态流的三个阶段一次完整的对话回合inner state 会经历三个阶段第一阶段读取当前状态Before state。这一步是为了给 LLM“提词”让它在生成回复前先了解已经发生的上下文。第二阶段生成回复与状态更新候选Generated state。LLM 在回复用户的同时也会输出一份建议的“内心独白更新”。通常我们会限制模型只更新部分字段避免它把已经确认的信息改掉。第三阶段提交新状态After state。Agent Core 校验并合并状态更新再写回存储。如果用户中途打断或话题切换状态更新可能不是单向追加而是需要基于会话策略做处理。3. 环境准备与项目结构3.1 运行环境说明本文的示例代码使用 Python 编写示例主要验证“inner monologue 状态管理”这一核心逻辑不依赖特定厂商的语音服务。你可以把它跑在本地开发机上。版本方面建议使用 Python 3.10 及以上版本。由于不同版本的 LLM SDK 差异较大本文不绑定具体 SDK 版本而是用一个抽象的call_llm()函数代替真实模型调用这样你可以很方便地替换成自己项目里的 OpenAI SDK、通义千问 SDK、文心 SDK 或本地部署模型接口。如果你希望在本地做最小验证只需要一个能跑 Python 的环境即可。如果要接入真实语音还需要准备ASR 服务将用户语音转成文本。TTS 服务将 Agent 回复文本转成语音。LLM 服务负责语言理解和生成。状态存储最简单的可以用内存字典生产环境建议使用 Redis 或数据库。3.2 依赖清单示例代码尽量使用标准库额外依赖很少依赖作用说明Python 3.10运行环境实际版本按项目调整redis-py生产环境状态存储本文示例先使用内存字典生产替换为 Redis任意 LLM SDK模型调用按实际服务商接入fastapi / flask可选用于上层接口封装本文不强制使用3.3 目录结构推荐把工程拆成下面这样的结构voice_agent_inner_monologue/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent 核心逻辑编排 LLM 与状态更新 │ ├── inner_state.py # inner monologue 数据结构与状态合并 │ ├── llm_client.py # 封装 LLM 调用 │ └── memory.py # 会话状态存储可替换为 Redis ├── prompt_templates.py # Prompt 模板 ├── demo.py # 命令行演示入口 └── README.md4. 核心实现给 Agent 加入内心独白这一节我们先从核心数据结构开始再逐步实现“读取状态 - 生成回复 - 更新状态”的闭环。4.1 定义内心独白的数据结构为了方便 LLM 理解inner monologue 最好使用 JSON 格式。我们会定义一个InnerState类用来承载task当前任务目标。facts已收集到的关键事实。pending还需要确认的问题。steps已经完成的操作步骤。preference用户偏好。history_summary一个高度压缩的对话摘要保存那些不适合放进字段但又有用的信息。# 文件路径agent/inner_state.py import json from typing import Any, Dict, List, Optional class InnerState: Inner Monologue 状态对象。 def __init__( self, task: str , facts: Optional[Dict[str, Any]] None, pending: Optional[List[str]] None, steps: Optional[List[str]] None, preference: Optional[Dict[str, Any]] None, history_summary: str , ) - None: self.task task self.facts facts or {} self.pending pending or [] self.steps steps or [] self.preference preference or {} self.history_summary history_summary def to_dict(self) - Dict[str, Any]: return { task: self.task, facts: self.facts, pending: self.pending, steps: self.steps, preference: self.preference, history_summary: self.history_summary, } def to_json(self) - str: return json.dumps(self.to_dict(), ensure_asciiFalse) classmethod def from_dict(cls, data: Dict[str, Any]) - InnerState: return cls( taskdata.get(task, ), factsdata.get(facts, {}), pendingdata.get(pending, []), stepsdata.get(steps, []), preferencedata.get(preference, {}), history_summarydata.get(history_summary, ), ) def merge_update(self, update: Dict[str, Any]) - None: 把 LLM 返回的状态更新合并进当前状态。 if task in update and update[task]: self.task update[task] if facts in update and isinstance(update[facts], dict): self.facts.update(update[facts]) if pending in update and isinstance(update[pending], list): self.pending update[pending] if steps in update and isinstance(update[steps], list): self.steps update[steps] if preference in update and isinstance(update[preference], dict): self.preference.update(update[preference]) if history_summary in update and update[history_summary]: self.history_summary update[history_summary]这里的关键是merge_update。它不会整体覆盖facts而是用dict.update合并更新这样上一轮确认的“靠窗”不会因为本轮没有提到就被清除。4.2 把内心独白注入 Prompt为了让 LLM 既能参考内心独白又能输出状态更新指令Prompt 模板要分两部分系统提示告诉 LLM 它的角色以及 inner monologue 的格式要求。用户输入部分包含当前 inner state 和本次用户语音转写文本。我建议在系统提示里明确两点必须基于 inner monologue 判断哪些信息已经收集不要重复追问。输出必须是一个 JSON 对象包含reply和state_update两个字段。# 文件路径prompt_templates.py SYSTEM_PROMPT 你是一个语音助手具备“内心独白”机制。 你的内心独白以 JSON 形式保存在独立字段中在每轮对话前会提供给你。 你必须严格按照以下要求工作 1. 先阅读 inner_state理解当前任务进度和已知信息。 2. 根据当前用户输入判断是否需要向用户追问缺失信息。 3. 如果某个信息已经在 inner_state.facts 中不要重复询问。 4. 回复要简短自然适合语音播放。 5. 你必须在输出中给出 reply 和 state_update 两个字段。 回复格式 { reply: 对用户说的话, state_update: { task: 当前任务, facts: {字段名: 字段值}, pending: [还需要确认的问题], steps: [已完成步骤], preference: {用户偏好: 偏好值}, history_summary: 压缩后的对话摘要 } } def build_user_prompt(user_text: str, inner_state_json: str, last_action_result: str ) - str: parts [ f【当前内心独白】\n{inner_state_json}, f【用户本次输入】\n{user_text}, ] if last_action_result: parts.insert(2, f【最近一次工具执行结果】\n{last_action_result}) return \n\n.join(parts)在真实工程中last_action_result可以承载“会议室查询接口返回的结果”“天气接口返回的数据”等这样 LLM 在生成回复时也能参考外部工具的结果。4.3 对话回合后的状态更新接下来实现AgentCore。一个完整的对话回合包括从内存中读取会话状态。构建 Prompt。调用 LLM。解析返回 JSON。提取reply和state_update。合并状态并保存。# 文件路径agent/core.py import json from typing import Dict, Any from agent.inner_state import InnerState from agent.llm_client import LLMClient from agent.memory import SessionMemory from prompt_templates import SYSTEM_PROMPT, build_user_prompt class AgentCore: def __init__(self, memory: SessionMemory, llm: LLMClient): self.memory memory self.llm llm def process_turn( self, session_id: str, user_text: str, last_action_result: str , ) - Dict[str, Any]: # 1. 读取当前状态 state self.memory.get(session_id) if state is None: state InnerState() # 2. 构建 Prompt user_prompt build_user_prompt( user_text, state.to_json(), last_action_result, ) # 3. 调用 LLM 并解析结果 raw_response self.llm.chat( system_promptSYSTEM_PROMPT, user_promptuser_prompt, ) parsed self._parse_response(raw_response) # 4. 合并状态更新 state.merge_update(parsed.get(state_update, {})) # 5. 保存状态 self.memory.set(session_id, state) return { reply: parsed.get(reply, ), inner_state: state.to_dict(), } staticmethod def _parse_response(raw_response: str) - Dict[str, Any]: 解析 LLM 返回兼容 JSON 包裹情况。 text raw_response.strip() if text.startswith(): text text.strip() if text.startswith(json): text text[4:].strip() try: return json.loads(text) except json.JSONDecodeError: # 这里可以加入重试或容错逻辑 return { reply: raw_response, state_update: {}, }这里_parse_response兼容了模型偶尔输出 Markdown 代码块的情况避免因为多了一个三个反引号而解析失败。4.4 处理语音识别的分句与打断真实语音场景里用户可能说一句话停一下然后继续说下一句。如果每停顿一次就触发一个回合状态会被频繁更新而且可能在用户还没说完时就插入回复体验很差。常用的做法是“端点检测VAD 静音超时”策略当检测到用户停顿超过阈值比如 600ms会把当前已识别的文本作为一句完整的用户输入。如果用户继续说话则累积文本并重置计时器。只有用户真正说完一句话才调用AgentCore.process_turn()。打断场景下用户可能会在 Agent 播报 A 任务时突然说“等等改成下午”。这时Agent 需要把当前播报内容标记为“未完成”然后重新进入回合读取当前 inner state把用户的新指令合并进去。伪代码如下# 文件路径agent/streaming_handler.py 示例思路 class StreamingVoiceHandler: def __init__(self, core: AgentCore): self.core core self.buffered_text def on_audio_chunk(self, session_id: str, is_end_of_sentence: bool, text: str) - str | None: self.buffered_text text if not is_end_of_sentence: return None # 继续等待 user_input self.buffered_text.strip() self.buffered_text if not user_input: return None # 把完整句子交给 AgentCore 处理 result self.core.process_turn(session_id, user_input) return result[reply]这里的is_end_of_sentence通常由 VAD 模块或 ASR 服务的“句末静音”事件提供。不同语音平台叫法不同但核心逻辑是一致的在一句话完整时再触发 Agent。5. 完整实战一个带记忆的语音 Agent 最小原型为了让读者能直接运行验证这一节实现一个不依赖真实语音和真实云端模型的最小原型。LLM 调用部分用本地模拟函数代替状态管理用内存字典核心逻辑与生产实现完全一致。5.1 创建状态管理模块先看memory.py# 文件路径agent/memory.py from typing import Dict, Optional from agent.inner_state import InnerState class SessionMemory: 基于内存的会话状态存储生产环境可替换为 Redis。 def __init__(self) - None: self._storage: Dict[str, InnerState] {} def get(self, session_id: str) - Optional[InnerState]: return self._storage.get(session_id) def set(self, session_id: str, state: InnerState) - None: self._storage[session_id] state5.2 编写 Agent 核心逻辑然后写一个模拟的LLMClient假装它能根据 inner state 生成回复和状态更新# 文件路径agent/llm_client.py import json from typing import Dict, Any class LLMClient: 模拟 LLM 客户端真实项目替换为实际 SDK 调用。 def chat(self, system_prompt: str, user_prompt: str) - str: # 这里可以替换成真实的模型调用 # response openai.chat.completions.create(...) # return response.choices[0].message.content # # 为了演示我们直接返回一份固定 JSON。 return json.dumps( { reply: 好的A302 会议室明天上午 10 点到 12 点我记住了。, state_update: { task: 预订会议室 A302, facts: { date: 明天, time_range: 10:00-12:00, room: A302 }, pending: [是否需要投影设备], steps: [查询空闲会议室, 确认 A302], preference: {window_seat: True}, history_summary: 用户希望订靠窗会议室。 } }, ensure_asciiFalse, )这个模拟返回值虽然不会真的理解语义但它能帮助你验证整个 inner monologue 状态流转机制是否正常。接入真实模型时只需要把chat方法内部替换为你的 SDK 调用即可。5.3 模拟语音输入运行最后写一个demo.py模拟连续两轮用户输入# 文件路径demo.py from agent.core import AgentCore from agent.llm_client import LLMClient from agent.memory import SessionMemory def main() - None: memory SessionMemory() llm LLMClient() core AgentCore(memorymemory, llmllm) session_id session-user-001 # 第一轮用户预订会议室 result core.process_turn( session_idsession_id, user_text帮我查一下明天上午 10 点到 12 点的空闲会议室要有靠窗的。, ) print(Agent 回复, result[reply]) print(Inner Monologue, result[inner_state]) print(- * 60) # 第二轮用户回问之前的信息 # 这里实际上不会再看完整历史而是从 inner_state 里读取已知信息 current_state memory.get(session_id) if current_state: room current_state.facts.get(room, 未知) time_range current_state.facts.get(time_range, 未知) date current_state.facts.get(date, 未知) print(f[模拟记忆检索] 已知会议室{room}时间{date} {time_range}) # 第三轮继续对话确认投影设备 result2 core.process_turn( session_idsession_id, user_text需要投影设备吗, ) print(Agent 回复, result2[reply]) print(Inner Monologue, result2[inner_state]) if __name__ __main__: main()运行命令python demo.py5.4 预期输出与效果说明运行后你会看到类似下面的输出Agent 回复 好的A302 会议室明天上午 10 点到 12 点我记住了。 Inner Monologue {task: 预订会议室 A302, facts: {date: 明天, time_range: 10:00-12:00, room: A302}, pending: [是否需要投影设备], steps: [查询空闲会议室, 确认 A302], preference: {window_seat: True}, history_summary: 用户希望订靠窗会议室。} ------------------------------------------------------------ [模拟记忆检索] 已知会议室A302时间明天 10:00-12:00 Agent 回复 好的A302 会议室明天上午 10 点到 12 点我记住了。 Inner Monologue {task: 预订会议室 A302, facts: {date: 明天, time_range: 10:00-12:00, room: A302}, pending: [是否需要投影设备], steps: [查询空闲会议室, 确认 A302], preference: {window_seat: True}, history_summary: 用户希望订靠窗会议室。}关键点在于第二轮即使你不重新给模型发历史对话Agent 也能从inner_state中拿到“会议室、时间”等关键信息。这就是 inner monologue 的价值把上下文从“原始文本流量”变成“结构化记忆”。6. 常见问题与排查思路在实际落地过程中inner monologue 机制可能不会第一次就跑通。下面总结几个高频问题。问题现象常见原因解决思路Agent 仍然重复询问已确认的信息state_update 没有正确合并 facts检查 merge_update 是否用 update 而不是整体覆盖状态字段越来越多Prompt 越来越长没有做字段裁剪与摘要压缩为 facts 设置上限历史摘要单独压缩模型返回的 JSON 解析失败模型输出被 Markdown 代码块包裹在解析前清理 和 json 前缀用户中途打断后状态错乱没有处理“未完成任务”的标记增加 interrupted 标记重新规划待办多用户并发访问同一会话状态内存字典没有锁使用 Redis 并带上版本号或事务更新状态更新丢失LLM 返回的 state_update 字段缺失解析失败时保留旧状态或增加校验重试Agent 在语音里回答太长没有限制 reply 长度Prompt 中明确要求 1 句话以内6.1 重复追问的排查这种现象通常是facts被错误覆盖导致的。比如上一轮写入facts[room] A302下一轮模型返回的state_update中没有room如果你的merge_update使用了“整体替换”而不是“字段合并”那room就会被清空。解决办法统一使用字典局部更新而不是整体覆盖。if facts in update and isinstance(update[facts], dict): self.facts.update(update[facts])6.2 状态膨胀与压缩如果每个字段都无限增长facts会变成一个垃圾桶。这时需要做两层处理第一层限制字段数量。比如规定facts最多保留 8 个关键字段超过后淘汰优先级最低的字段。第二层使用history_summary。让模型在每轮更新时把不再作为结构化字段保存的信息压缩成一句摘要。例如{ history_summary: 用户之前提过靠窗偏好后来改口说靠门也可以。 }这样既保留了长期记忆又不会让 Payload 无限膨胀。6.3 并发写冲突语音场景下同一个用户可能通过多个设备登录也可能 Agent 内部的异步任务与对话任务同时写状态。此时不能简单使用内存字典。生产环境建议使用 Redis Hash 存储状态。为状态保存增加版本号每次更新时检查版本。更新失败时重试或丢弃过期写入。# 伪代码带版本号的状态写入 state_key fvoice_agent:{session_id}:state version redis.hget(state_key, version) new_version int(version) 1 # 使用 Lua 脚本或 WATCH 事务保证原子性6.4 模型输出不稳定大模型不一定每次都输出严格 JSON尤其是没有做响应格式化约束的模型。建议在 Prompt 中给出一个 JSON 示例。开启服务端 JSON Mode如果服务商支持。解析失败时保留上一份状态不要让 Agent 直接崩溃。对 reply 字段单独降级处理如果解析失败就把模型原始输出当作 reply同时不更新状态。7. 工程落地建议与最佳实践7.1 状态设计原则Inner monologue 的设计目标是“最少的字段记录最关键的决策”不要试图把用户说的每个字都变成结构化字段。字段建议严格分类facts用户明确给出的硬信息如时间、地点、人名、数量。pending还需要追问的问题每次最多 3~5 个。steps已完成的操作步骤主要用于回溯和排错。preference用户偏好需要跨任务保留。history_summary非结构化摘要用来兜底。需要特别注意不要把facts长期无差别累积。当会话目标完成或用户发起新任务时要执行一次“状态清理”把上一个任务的facts压缩进history_summary。7.2 持久化与容灾内存字典只适合开发调试。生产环境至少做到用 Redis 保存 inner state设置会话过期时间。关键任务状态异步持久化到数据库防止服务重启丢失。为状态变更保留日志方便问题回溯。语音数据和转写文本满足数据合规要求该脱敏的脱敏。7.3 隐私与安全语音 Agent 会接触大量用户隐私信息包括声纹、位置、日程、支付意图等。工程上有几个底线inner state 中不要存储明文密码、支付密钥等敏感信息。会话数据默认加密存储过期自动清理。在用户授权范围内使用对话记录做模型优化不能私下留存音频。涉及下单、支付、删除等高风险操作前必须二次确认。7.4 模型选型与 Prompt 优化你可能会听到“某个模型很聪明一定不会忘”的说法。有能力的长上下文模型确实能缓解遗忘但工程上依然建议保留 inner monologue。原因很简单长上下文模型既贵又慢而语音交互又是高频、低延迟场景。用 inner monologue 把关键信息抽出来再配合短 Prompt可以在成本和体验之间取得更好的平衡。在 Prompt 优化方面几条经验值得参考给 LLM 明确的“先读内心独白再回复”的指令。明确禁止重复询问已确认信息。让回复保持一句话以内适合 TTS 播放。对state_update给出字段说明避免模型自由发挥。在模型返回多轮更新时优先处理最后一份更新。7.5 可观测性建设调试 voice agent 比调试普通接口更难因为音频链路长、状态流转多。建议为每个会话保留结构化日志{ session_id: session-user-001, turn_id: 1, user_text: 帮我查一下明天上午 10 点到 12 点的空闲会议室, inner_state_before: {...}, llm_response: ..., inner_state_after: {...}, latency_ms: 480, reply_text: ... }有了这类日志当用户反馈“Agent 忘了我刚才说过的话”时可以直接查看inner_state_before是否已经包含该信息快速定位是状态更新丢失、LLM 输出异常、还是状态读取延迟。8. 总结与下一步Inner monologue 的提出本质上是把语音 Agent 从一个“无状态翻译机”升级为“有状态任务助手”。它通过显式的内部状态记录任务目标、已知信息、待确认项和用户偏好让 Agent 在没有完整历史文本的情况下也能稳定回应用户。本文从背景、架构、数据结构、核心代码到工程实践完整演示了如何给 voice agent 加上这套机制。你可以基于这套最小原型继续扩展接入真实 ASR/TTS 服务验证真实语音链路。把内存 SessionMemory 替换为 Redis支持多实例部署。让LLMClient接入你实际使用的模型 API并在 Prompt 中加入更多业务约束。增加工具调用能力让 Agent 能查会议室、订日程、点亮灯等。如果你正在做语音助手或 Agent 类产品建议先从“用户最容易感知遗忘”的场景入手比如订会议室、查快递、点外卖这类多轮任务。先让 inner monologue 跑通一个典型流程再逐步覆盖更多场景会比一开始就做一套大而全的状态系统更稳妥。 语音Agent的内心独白：上下文记忆机制原理与工程落地
语音交互场景下“说完就忘”简直是语音 Agent 落地过程中最让人头疼的问题。用户明明在一分钟前说过“帮我订靠窗的位置”等 Agent 确认时间时却表现得像第一次听到一样。本文围绕 voice agent 的上下文遗忘问题完整拆解一种“内心独白inner monologue”机制它是如何工作的、如何设计状态结构、怎么和 ASR / LLM / TTS 串联以及落地时有哪些坑。全程包含结构设计、可运行示例代码、常见问题排查和工程建议适合正在做语音助手、Agent 应用或对话系统的开发者。1. 背景语音 Agent 为什么会“忘事”1.1 多轮语音对话的上下文困境先看一个很自然的用户交互用户帮我查一下明天上午 10 点到 12 点的空闲会议室。 Agent好的我找到了 3 间A302 靠窗B101 有投影C205 空着。 用户那就订 A302 吧。 Agent好的需要我同时帮你预定投影设备吗 用户等下刚才那间靠窗的会议室是明天几点来着如果 Agent 在这一轮“失忆”了用户体验会非常割裂。语音 Agent 和文字聊天机器人不同它有以下几个天然难点会话连续性弱用户可能中途打断、换话题、很久之后又切回前面的任务模型很难知道当前该以哪条对话主线为准。上下文窗口有限即便模型支持很长的上下文把每一轮语音转写文本全部塞进 Prompt也会导致成本升高、响应变慢甚至把关键信息淹没在噪音里。语音转写有噪声ASR 可能出错一句“靠窗”可能被识别成“靠场”如果 Agent 只依赖原始转写文本很容易被带偏。交互节奏快语音场景讲究低延迟不能像文字对话那样让用户等很久。说到底语音 Agent 缺的不是“模型能力”而是一个自己能随时查看、能持续更新的“工作记忆”。这正是本文要讲的 inner monologue 机制可以解决的问题。1.2 Inner Monologue 是什么Inner Monologue直译是“内心独白”。在普通对话里一个人在回答别人之前通常会在心里快速整理一下对方刚才想干什么 我已经确认了哪些信息 还差什么信息 我现在应该问什么语音 Agent 的 inner monologue 就是把这个“心里整理”的过程显式建模出来。它是在用户输入和 Agent 输出之间插入的一段结构化内部状态记录当前任务的目标、已收集的信息、待确认的细节、历史关键决策等。与直接拼接“完整聊天历史”不同inner monologue 是一种“压缩后”的记忆。它不保存每一帧语音文本而是保存这个 Agent 在当前会话中真正需要记住的事实。举个例子用户订会议室这个场景下Agent 的内心独白可能长这样{ current_task: 预订会议室 A302, known_info: { date: 明天, time_range: 10:00-12:00, room: A302 }, pending_questions: [是否需要投影设备], session_goal: 完成会议室预订, user_preference: { window_seat: true } }当用户问“刚才那间靠窗的会议室是明天几点来着”时Agent 不需要翻历史记录直接查 inner monologue 里的known_info就能回答“A302明天上午 10 点到 12 点”。1.3 为什么不是“把所有历史都拼进 Prompt”有的开发者会问“直接把前 N 轮对话全部拼进 Prompt 不就行了吗”这种做法在短时间内有用但长期来看问题不少成本爆炸每多一轮Token 消耗就增加一部分语音场景高频调用费用增长很快。延迟升高Prompt 越长模型生成首字时间越晚用户会觉得“AI 反应变慢”。信息稀释关键信息淹没在一堆“嗯”“然后”“那个”等口语词里模型反而提取不准。超出窗口对话一旦变长最终会顶到上下文上限必须做裁剪而简单裁剪会丢失重要信息。Inner monologue 的核心思路是“先提炼再记忆”。它把“记住什么”这件事交给一个显式的状态管理模块去完成而不是把所有原始数据一股脑丢给模型。这样模型每次只需要读一份紧凑的内部状态外加当前用户输入就能稳定地产出高质量回复。2. 整体架构设计2.1 一条语音请求的处理链路一个带 inner monologue 的 voice agent其处理链路大致如下用户说话 - ASR语音转文字 - Agent Core更新 inner monologue决定回复内容 - LLM根据内心独白生成回复文本 - TTS文本转语音 - 用户听到回复这里的核心是Agent Core它负责两部分工作在每轮对话开始前把当前的 inner monologue 和本次用户输入一起交给 LLM。在每轮对话结束后根据 LLM 的输出和用户输入更新 inner monologue。如果把Agent Core拆细流程可以写成接收 ASR 输出的文本。读取当前会话的 inner state。把 inner state 当前输入 任务提示词一起发送给 LLM。从 LLM 返回结果中解析两个部分reply对用户说的话和state_update对内心独白的更新指令。执行状态更新写入新的 inner state。把reply送入 TTS。2.2 Inner Monologue 模块的职责Inner Monologue 模块并不是一个独立的大模型服务它更像一个“状态机 记忆数据库”。它负责维护 Session 级状态每个用户会话对应一份独立的 inner state。解析意图与信息槽位从用户输入中提取关键实体。维护待确认问题列表哪些信息还没拿到。记录已完成步骤任务执行到哪一步。处理状态回滚用户反悔、修改需求时能回到之前的状态。2.3 状态流的三个阶段一次完整的对话回合inner state 会经历三个阶段第一阶段读取当前状态Before state。这一步是为了给 LLM“提词”让它在生成回复前先了解已经发生的上下文。第二阶段生成回复与状态更新候选Generated state。LLM 在回复用户的同时也会输出一份建议的“内心独白更新”。通常我们会限制模型只更新部分字段避免它把已经确认的信息改掉。第三阶段提交新状态After state。Agent Core 校验并合并状态更新再写回存储。如果用户中途打断或话题切换状态更新可能不是单向追加而是需要基于会话策略做处理。3. 环境准备与项目结构3.1 运行环境说明本文的示例代码使用 Python 编写示例主要验证“inner monologue 状态管理”这一核心逻辑不依赖特定厂商的语音服务。你可以把它跑在本地开发机上。版本方面建议使用 Python 3.10 及以上版本。由于不同版本的 LLM SDK 差异较大本文不绑定具体 SDK 版本而是用一个抽象的call_llm()函数代替真实模型调用这样你可以很方便地替换成自己项目里的 OpenAI SDK、通义千问 SDK、文心 SDK 或本地部署模型接口。如果你希望在本地做最小验证只需要一个能跑 Python 的环境即可。如果要接入真实语音还需要准备ASR 服务将用户语音转成文本。TTS 服务将 Agent 回复文本转成语音。LLM 服务负责语言理解和生成。状态存储最简单的可以用内存字典生产环境建议使用 Redis 或数据库。3.2 依赖清单示例代码尽量使用标准库额外依赖很少依赖作用说明Python 3.10运行环境实际版本按项目调整redis-py生产环境状态存储本文示例先使用内存字典生产替换为 Redis任意 LLM SDK模型调用按实际服务商接入fastapi / flask可选用于上层接口封装本文不强制使用3.3 目录结构推荐把工程拆成下面这样的结构voice_agent_inner_monologue/ ├── agent/ │ ├── __init__.py │ ├── core.py # Agent 核心逻辑编排 LLM 与状态更新 │ ├── inner_state.py # inner monologue 数据结构与状态合并 │ ├── llm_client.py # 封装 LLM 调用 │ └── memory.py # 会话状态存储可替换为 Redis ├── prompt_templates.py # Prompt 模板 ├── demo.py # 命令行演示入口 └── README.md4. 核心实现给 Agent 加入内心独白这一节我们先从核心数据结构开始再逐步实现“读取状态 - 生成回复 - 更新状态”的闭环。4.1 定义内心独白的数据结构为了方便 LLM 理解inner monologue 最好使用 JSON 格式。我们会定义一个InnerState类用来承载task当前任务目标。facts已收集到的关键事实。pending还需要确认的问题。steps已经完成的操作步骤。preference用户偏好。history_summary一个高度压缩的对话摘要保存那些不适合放进字段但又有用的信息。# 文件路径agent/inner_state.py import json from typing import Any, Dict, List, Optional class InnerState: Inner Monologue 状态对象。 def __init__( self, task: str , facts: Optional[Dict[str, Any]] None, pending: Optional[List[str]] None, steps: Optional[List[str]] None, preference: Optional[Dict[str, Any]] None, history_summary: str , ) - None: self.task task self.facts facts or {} self.pending pending or [] self.steps steps or [] self.preference preference or {} self.history_summary history_summary def to_dict(self) - Dict[str, Any]: return { task: self.task, facts: self.facts, pending: self.pending, steps: self.steps, preference: self.preference, history_summary: self.history_summary, } def to_json(self) - str: return json.dumps(self.to_dict(), ensure_asciiFalse) classmethod def from_dict(cls, data: Dict[str, Any]) - InnerState: return cls( taskdata.get(task, ), factsdata.get(facts, {}), pendingdata.get(pending, []), stepsdata.get(steps, []), preferencedata.get(preference, {}), history_summarydata.get(history_summary, ), ) def merge_update(self, update: Dict[str, Any]) - None: 把 LLM 返回的状态更新合并进当前状态。 if task in update and update[task]: self.task update[task] if facts in update and isinstance(update[facts], dict): self.facts.update(update[facts]) if pending in update and isinstance(update[pending], list): self.pending update[pending] if steps in update and isinstance(update[steps], list): self.steps update[steps] if preference in update and isinstance(update[preference], dict): self.preference.update(update[preference]) if history_summary in update and update[history_summary]: self.history_summary update[history_summary]这里的关键是merge_update。它不会整体覆盖facts而是用dict.update合并更新这样上一轮确认的“靠窗”不会因为本轮没有提到就被清除。4.2 把内心独白注入 Prompt为了让 LLM 既能参考内心独白又能输出状态更新指令Prompt 模板要分两部分系统提示告诉 LLM 它的角色以及 inner monologue 的格式要求。用户输入部分包含当前 inner state 和本次用户语音转写文本。我建议在系统提示里明确两点必须基于 inner monologue 判断哪些信息已经收集不要重复追问。输出必须是一个 JSON 对象包含reply和state_update两个字段。# 文件路径prompt_templates.py SYSTEM_PROMPT 你是一个语音助手具备“内心独白”机制。 你的内心独白以 JSON 形式保存在独立字段中在每轮对话前会提供给你。 你必须严格按照以下要求工作 1. 先阅读 inner_state理解当前任务进度和已知信息。 2. 根据当前用户输入判断是否需要向用户追问缺失信息。 3. 如果某个信息已经在 inner_state.facts 中不要重复询问。 4. 回复要简短自然适合语音播放。 5. 你必须在输出中给出 reply 和 state_update 两个字段。 回复格式 { reply: 对用户说的话, state_update: { task: 当前任务, facts: {字段名: 字段值}, pending: [还需要确认的问题], steps: [已完成步骤], preference: {用户偏好: 偏好值}, history_summary: 压缩后的对话摘要 } } def build_user_prompt(user_text: str, inner_state_json: str, last_action_result: str ) - str: parts [ f【当前内心独白】\n{inner_state_json}, f【用户本次输入】\n{user_text}, ] if last_action_result: parts.insert(2, f【最近一次工具执行结果】\n{last_action_result}) return \n\n.join(parts)在真实工程中last_action_result可以承载“会议室查询接口返回的结果”“天气接口返回的数据”等这样 LLM 在生成回复时也能参考外部工具的结果。4.3 对话回合后的状态更新接下来实现AgentCore。一个完整的对话回合包括从内存中读取会话状态。构建 Prompt。调用 LLM。解析返回 JSON。提取reply和state_update。合并状态并保存。# 文件路径agent/core.py import json from typing import Dict, Any from agent.inner_state import InnerState from agent.llm_client import LLMClient from agent.memory import SessionMemory from prompt_templates import SYSTEM_PROMPT, build_user_prompt class AgentCore: def __init__(self, memory: SessionMemory, llm: LLMClient): self.memory memory self.llm llm def process_turn( self, session_id: str, user_text: str, last_action_result: str , ) - Dict[str, Any]: # 1. 读取当前状态 state self.memory.get(session_id) if state is None: state InnerState() # 2. 构建 Prompt user_prompt build_user_prompt( user_text, state.to_json(), last_action_result, ) # 3. 调用 LLM 并解析结果 raw_response self.llm.chat( system_promptSYSTEM_PROMPT, user_promptuser_prompt, ) parsed self._parse_response(raw_response) # 4. 合并状态更新 state.merge_update(parsed.get(state_update, {})) # 5. 保存状态 self.memory.set(session_id, state) return { reply: parsed.get(reply, ), inner_state: state.to_dict(), } staticmethod def _parse_response(raw_response: str) - Dict[str, Any]: 解析 LLM 返回兼容 JSON 包裹情况。 text raw_response.strip() if text.startswith(): text text.strip() if text.startswith(json): text text[4:].strip() try: return json.loads(text) except json.JSONDecodeError: # 这里可以加入重试或容错逻辑 return { reply: raw_response, state_update: {}, }这里_parse_response兼容了模型偶尔输出 Markdown 代码块的情况避免因为多了一个三个反引号而解析失败。4.4 处理语音识别的分句与打断真实语音场景里用户可能说一句话停一下然后继续说下一句。如果每停顿一次就触发一个回合状态会被频繁更新而且可能在用户还没说完时就插入回复体验很差。常用的做法是“端点检测VAD 静音超时”策略当检测到用户停顿超过阈值比如 600ms会把当前已识别的文本作为一句完整的用户输入。如果用户继续说话则累积文本并重置计时器。只有用户真正说完一句话才调用AgentCore.process_turn()。打断场景下用户可能会在 Agent 播报 A 任务时突然说“等等改成下午”。这时Agent 需要把当前播报内容标记为“未完成”然后重新进入回合读取当前 inner state把用户的新指令合并进去。伪代码如下# 文件路径agent/streaming_handler.py 示例思路 class StreamingVoiceHandler: def __init__(self, core: AgentCore): self.core core self.buffered_text def on_audio_chunk(self, session_id: str, is_end_of_sentence: bool, text: str) - str | None: self.buffered_text text if not is_end_of_sentence: return None # 继续等待 user_input self.buffered_text.strip() self.buffered_text if not user_input: return None # 把完整句子交给 AgentCore 处理 result self.core.process_turn(session_id, user_input) return result[reply]这里的is_end_of_sentence通常由 VAD 模块或 ASR 服务的“句末静音”事件提供。不同语音平台叫法不同但核心逻辑是一致的在一句话完整时再触发 Agent。5. 完整实战一个带记忆的语音 Agent 最小原型为了让读者能直接运行验证这一节实现一个不依赖真实语音和真实云端模型的最小原型。LLM 调用部分用本地模拟函数代替状态管理用内存字典核心逻辑与生产实现完全一致。5.1 创建状态管理模块先看memory.py# 文件路径agent/memory.py from typing import Dict, Optional from agent.inner_state import InnerState class SessionMemory: 基于内存的会话状态存储生产环境可替换为 Redis。 def __init__(self) - None: self._storage: Dict[str, InnerState] {} def get(self, session_id: str) - Optional[InnerState]: return self._storage.get(session_id) def set(self, session_id: str, state: InnerState) - None: self._storage[session_id] state5.2 编写 Agent 核心逻辑然后写一个模拟的LLMClient假装它能根据 inner state 生成回复和状态更新# 文件路径agent/llm_client.py import json from typing import Dict, Any class LLMClient: 模拟 LLM 客户端真实项目替换为实际 SDK 调用。 def chat(self, system_prompt: str, user_prompt: str) - str: # 这里可以替换成真实的模型调用 # response openai.chat.completions.create(...) # return response.choices[0].message.content # # 为了演示我们直接返回一份固定 JSON。 return json.dumps( { reply: 好的A302 会议室明天上午 10 点到 12 点我记住了。, state_update: { task: 预订会议室 A302, facts: { date: 明天, time_range: 10:00-12:00, room: A302 }, pending: [是否需要投影设备], steps: [查询空闲会议室, 确认 A302], preference: {window_seat: True}, history_summary: 用户希望订靠窗会议室。 } }, ensure_asciiFalse, )这个模拟返回值虽然不会真的理解语义但它能帮助你验证整个 inner monologue 状态流转机制是否正常。接入真实模型时只需要把chat方法内部替换为你的 SDK 调用即可。5.3 模拟语音输入运行最后写一个demo.py模拟连续两轮用户输入# 文件路径demo.py from agent.core import AgentCore from agent.llm_client import LLMClient from agent.memory import SessionMemory def main() - None: memory SessionMemory() llm LLMClient() core AgentCore(memorymemory, llmllm) session_id session-user-001 # 第一轮用户预订会议室 result core.process_turn( session_idsession_id, user_text帮我查一下明天上午 10 点到 12 点的空闲会议室要有靠窗的。, ) print(Agent 回复, result[reply]) print(Inner Monologue, result[inner_state]) print(- * 60) # 第二轮用户回问之前的信息 # 这里实际上不会再看完整历史而是从 inner_state 里读取已知信息 current_state memory.get(session_id) if current_state: room current_state.facts.get(room, 未知) time_range current_state.facts.get(time_range, 未知) date current_state.facts.get(date, 未知) print(f[模拟记忆检索] 已知会议室{room}时间{date} {time_range}) # 第三轮继续对话确认投影设备 result2 core.process_turn( session_idsession_id, user_text需要投影设备吗, ) print(Agent 回复, result2[reply]) print(Inner Monologue, result2[inner_state]) if __name__ __main__: main()运行命令python demo.py5.4 预期输出与效果说明运行后你会看到类似下面的输出Agent 回复 好的A302 会议室明天上午 10 点到 12 点我记住了。 Inner Monologue {task: 预订会议室 A302, facts: {date: 明天, time_range: 10:00-12:00, room: A302}, pending: [是否需要投影设备], steps: [查询空闲会议室, 确认 A302], preference: {window_seat: True}, history_summary: 用户希望订靠窗会议室。} ------------------------------------------------------------ [模拟记忆检索] 已知会议室A302时间明天 10:00-12:00 Agent 回复 好的A302 会议室明天上午 10 点到 12 点我记住了。 Inner Monologue {task: 预订会议室 A302, facts: {date: 明天, time_range: 10:00-12:00, room: A302}, pending: [是否需要投影设备], steps: [查询空闲会议室, 确认 A302], preference: {window_seat: True}, history_summary: 用户希望订靠窗会议室。}关键点在于第二轮即使你不重新给模型发历史对话Agent 也能从inner_state中拿到“会议室、时间”等关键信息。这就是 inner monologue 的价值把上下文从“原始文本流量”变成“结构化记忆”。6. 常见问题与排查思路在实际落地过程中inner monologue 机制可能不会第一次就跑通。下面总结几个高频问题。问题现象常见原因解决思路Agent 仍然重复询问已确认的信息state_update 没有正确合并 facts检查 merge_update 是否用 update 而不是整体覆盖状态字段越来越多Prompt 越来越长没有做字段裁剪与摘要压缩为 facts 设置上限历史摘要单独压缩模型返回的 JSON 解析失败模型输出被 Markdown 代码块包裹在解析前清理 和 json 前缀用户中途打断后状态错乱没有处理“未完成任务”的标记增加 interrupted 标记重新规划待办多用户并发访问同一会话状态内存字典没有锁使用 Redis 并带上版本号或事务更新状态更新丢失LLM 返回的 state_update 字段缺失解析失败时保留旧状态或增加校验重试Agent 在语音里回答太长没有限制 reply 长度Prompt 中明确要求 1 句话以内6.1 重复追问的排查这种现象通常是facts被错误覆盖导致的。比如上一轮写入facts[room] A302下一轮模型返回的state_update中没有room如果你的merge_update使用了“整体替换”而不是“字段合并”那room就会被清空。解决办法统一使用字典局部更新而不是整体覆盖。if facts in update and isinstance(update[facts], dict): self.facts.update(update[facts])6.2 状态膨胀与压缩如果每个字段都无限增长facts会变成一个垃圾桶。这时需要做两层处理第一层限制字段数量。比如规定facts最多保留 8 个关键字段超过后淘汰优先级最低的字段。第二层使用history_summary。让模型在每轮更新时把不再作为结构化字段保存的信息压缩成一句摘要。例如{ history_summary: 用户之前提过靠窗偏好后来改口说靠门也可以。 }这样既保留了长期记忆又不会让 Payload 无限膨胀。6.3 并发写冲突语音场景下同一个用户可能通过多个设备登录也可能 Agent 内部的异步任务与对话任务同时写状态。此时不能简单使用内存字典。生产环境建议使用 Redis Hash 存储状态。为状态保存增加版本号每次更新时检查版本。更新失败时重试或丢弃过期写入。# 伪代码带版本号的状态写入 state_key fvoice_agent:{session_id}:state version redis.hget(state_key, version) new_version int(version) 1 # 使用 Lua 脚本或 WATCH 事务保证原子性6.4 模型输出不稳定大模型不一定每次都输出严格 JSON尤其是没有做响应格式化约束的模型。建议在 Prompt 中给出一个 JSON 示例。开启服务端 JSON Mode如果服务商支持。解析失败时保留上一份状态不要让 Agent 直接崩溃。对 reply 字段单独降级处理如果解析失败就把模型原始输出当作 reply同时不更新状态。7. 工程落地建议与最佳实践7.1 状态设计原则Inner monologue 的设计目标是“最少的字段记录最关键的决策”不要试图把用户说的每个字都变成结构化字段。字段建议严格分类facts用户明确给出的硬信息如时间、地点、人名、数量。pending还需要追问的问题每次最多 3~5 个。steps已完成的操作步骤主要用于回溯和排错。preference用户偏好需要跨任务保留。history_summary非结构化摘要用来兜底。需要特别注意不要把facts长期无差别累积。当会话目标完成或用户发起新任务时要执行一次“状态清理”把上一个任务的facts压缩进history_summary。7.2 持久化与容灾内存字典只适合开发调试。生产环境至少做到用 Redis 保存 inner state设置会话过期时间。关键任务状态异步持久化到数据库防止服务重启丢失。为状态变更保留日志方便问题回溯。语音数据和转写文本满足数据合规要求该脱敏的脱敏。7.3 隐私与安全语音 Agent 会接触大量用户隐私信息包括声纹、位置、日程、支付意图等。工程上有几个底线inner state 中不要存储明文密码、支付密钥等敏感信息。会话数据默认加密存储过期自动清理。在用户授权范围内使用对话记录做模型优化不能私下留存音频。涉及下单、支付、删除等高风险操作前必须二次确认。7.4 模型选型与 Prompt 优化你可能会听到“某个模型很聪明一定不会忘”的说法。有能力的长上下文模型确实能缓解遗忘但工程上依然建议保留 inner monologue。原因很简单长上下文模型既贵又慢而语音交互又是高频、低延迟场景。用 inner monologue 把关键信息抽出来再配合短 Prompt可以在成本和体验之间取得更好的平衡。在 Prompt 优化方面几条经验值得参考给 LLM 明确的“先读内心独白再回复”的指令。明确禁止重复询问已确认信息。让回复保持一句话以内适合 TTS 播放。对state_update给出字段说明避免模型自由发挥。在模型返回多轮更新时优先处理最后一份更新。7.5 可观测性建设调试 voice agent 比调试普通接口更难因为音频链路长、状态流转多。建议为每个会话保留结构化日志{ session_id: session-user-001, turn_id: 1, user_text: 帮我查一下明天上午 10 点到 12 点的空闲会议室, inner_state_before: {...}, llm_response: ..., inner_state_after: {...}, latency_ms: 480, reply_text: ... }有了这类日志当用户反馈“Agent 忘了我刚才说过的话”时可以直接查看inner_state_before是否已经包含该信息快速定位是状态更新丢失、LLM 输出异常、还是状态读取延迟。8. 总结与下一步Inner monologue 的提出本质上是把语音 Agent 从一个“无状态翻译机”升级为“有状态任务助手”。它通过显式的内部状态记录任务目标、已知信息、待确认项和用户偏好让 Agent 在没有完整历史文本的情况下也能稳定回应用户。本文从背景、架构、数据结构、核心代码到工程实践完整演示了如何给 voice agent 加上这套机制。你可以基于这套最小原型继续扩展接入真实 ASR/TTS 服务验证真实语音链路。把内存 SessionMemory 替换为 Redis支持多实例部署。让LLMClient接入你实际使用的模型 API并在 Prompt 中加入更多业务约束。增加工具调用能力让 Agent 能查会议室、订日程、点亮灯等。如果你正在做语音助手或 Agent 类产品建议先从“用户最容易感知遗忘”的场景入手比如订会议室、查快递、点外卖这类多轮任务。先让 inner monologue 跑通一个典型流程再逐步覆盖更多场景会比一开始就做一套大而全的状态系统更稳妥。