最近在尝试构建自己的 AI Agent 应用时发现很多框架要么过于复杂要么对运行过程的控制不够透明。直到深入研究了 DeepSeek Harness 的源码才真正理解了 Agent 从启动、执行到状态管理的完整生命周期。本文将带你从源码层面拆解 DeepSeek Harness 的核心运行机制特别是 Agent、Turn、Step 的协作关系以及 Session 重建这一关键容错策略。无论你是想深入理解 Agent 框架设计还是希望在自己的项目中实现类似的能力这篇文章都能提供一套完整的代码级分析思路。1. 背景与核心概念在深入源码之前我们有必要先厘清几个核心概念这有助于我们理解 DeepSeek Harness 的设计哲学和要解决的根本问题。1.1 什么是 DeepSeek HarnessDeepSeek Harness 是一个用于构建、管理和执行 AI Agent 的框架。你可以把它想象成一个“马具”Harness它的核心作用是将强大的大语言模型如 DeepSeek 系列模型“套”起来通过一套标准化的流程和控制机制让模型能够稳定、可靠地完成复杂的、多步骤的任务。它不是一个具体的 Agent而是一个生产 Agent 的“工厂”和“调度中心”。1.2 Agent、Turn、Step 与 Session 的关系这是理解整个框架运行机制的关键。我们可以用一个“对话式任务执行”的类比来理解Agent代理这是任务执行的主体。它被赋予一个目标Goal并拥有工具Tools、记忆Memory等能力。Agent 本身是一个相对静态的“配置实体”它定义了“谁”来执行任务以及“用什么”来执行。Session会话这是 Agent一次完整的任务执行过程。当你启动一个 Agent 去完成一个目标时就创建了一个 Session。Session 封装了这次任务执行的全部上下文包括初始目标、与环境的交互历史、当前状态等。它是动态的、有生命周期的。Turn轮次这是 Session 内部的一个宏观执行单元。通常Agent 完成一个目标需要多次“思考-行动-观察”的循环。每一次完整的循环可以看作一个 Turn。例如Agent 先“思考”需要搜索天气然后“执行”搜索工具最后“观察”搜索结果这构成一个 Turn。Turn 是面向业务逻辑的。Step步骤这是 Turn 内部更细粒度的执行步骤是框架与底层模型交互的最小单元。一个 Turn 通常由多个 Step 组成例如1) 生成推理链的 Step2) 调用工具的 Step3) 处理工具返回结果的 Step。Step 是面向框架执行引擎的。它们的关系可以概括为一个Agent启动后会创建一个Session来追踪本次任务。在该 Session 中Agent 通过完成多个Turn来逐步推进任务。而每一个 Turn则由框架拆解为一系列有序的Step来具体执行。1.3 为什么需要 Session 重建这是分布式、长耗时 Agent 系统中的一个经典问题。想象一下你的 Agent 任务可能需要运行几分钟甚至几小时在这期间服务可能因为部署、故障或负载均衡而重启。如果 Session 状态完全存在于内存中那么重启将导致任务丢失一切从头开始。Session 重建Session Recovery/Reconstruction就是为了解决这个问题。其核心思想是将 Session 的运行状态包括历史 Turns、Steps、工具调用结果、中间思考过程等持久化到外部存储如数据库、Redis。当服务中断后重启框架能够根据一个唯一的 Session ID从持久化存储中加载出完整的历史状态并让 Agent 从“中断点”继续执行而不是重新开始。这为生产环境提供了至关重要的可靠性和容错性。2. 环境准备与源码定位为了进行源码解析我们需要先准备好环境。本文的分析基于 DeepSeek Harness 的公开源码假设版本为v0.1.x左右具体 commit 可能变化但核心架构相对稳定。1. 获取源码git clone https://github.com/deepseek-ai/deepseek-harness.git cd deepseek-harness建议查看README.md和CONTRIBUTING.md了解项目结构和基本要求。2. 主要目录结构分析重点deepseek-harness/ ├── src/ │ ├── harness/ # 框架核心 │ │ ├── agent/ # Agent 定义、配置、构建器 │ │ ├── session/ # Session 生命周期管理、状态存储、重建逻辑 │ │ ├── turn/ # Turn 的执行流程控制 │ │ ├── step/ # Step 的定义与执行器 │ │ ├── engine/ # 执行引擎协调 Agent/Session/Turn/Step │ │ ├── memory/ # 记忆模块可能与会话状态结合 │ │ ├── tools/ # 工具系统 │ │ └── llm/ # 大语言模型集成层 │ └── common/ # 通用工具、常量、异常 ├── examples/ # 示例代码 ├── tests/ # 单元测试和集成测试 └── requirements.txt # Python 依赖3. 推荐的分析工具IDE使用 VS Code 或 PyCharm它们能提供优秀的代码导航和跳转。搜索在 IDE 中全局搜索关键词如SessionRecovery、reconstruct_session、persist_state、TurnExecutor、StepRunner。调试虽然不运行但可以查看examples/目录下的示例理解 API 的调用方式。版本说明Agent 框架发展迅速本文聚焦于核心设计模式和运行机制这些思想具有普适性。具体类名、方法签名可能随版本迭代而变化但分析思路是相通的。阅读时请关注设计思想而非绝对路径。3. 核心运行机制源码拆解现在我们深入到源码内部看看 Agent 是如何一步步动起来的。3.1 Agent 的初始化与配置Agent 的创建通常通过一个构建器Builder或配置类来完成。我们可以在src/harness/agent/目录下找到相关代码。核心类分析AgentConfig这是一个数据类可能使用 Pydantic定义了 Agent 的静态属性。# 示例性代码反映核心字段 class AgentConfig: name: str model: str # 使用的 LLM如 “deepseek-chat” system_prompt: str # 系统指令定义 Agent 角色和能力 tools: List[ToolDefinition] # 可用的工具列表 max_turns: Optional[int] 10 # 最大 Turn 数防止无限循环 memory_config: Optional[MemoryConfig] None # 记忆配置AgentBuilder或AgentFactory负责根据AgentConfig组装出可运行的Agent实例。这个过程会初始化 LLM 客户端、注册工具、设置记忆系统等。class AgentBuilder: def __init__(self, config: AgentConfig): self.config config self.llm_client self._init_llm_client() self.tool_registry self._init_tool_registry() def build(self) - Agent: 构建一个 Agent 实例。 return Agent( configself.config, llm_clientself.llm_client, tool_registryself.tool_registry, # ... 其他依赖注入 )关键点Agent对象本身通常不包含运行时状态那是 Session 的职责它更像是一个“蓝图”或“模板”。3.2 Session 的生命周期管理Session 是运行时状态的核心容器。相关代码主要在src/harness/session/。核心类分析Session类这是最重要的类之一。它持有一次任务执行的所有上下文。class Session: def __init__(self, session_id: str, agent: Agent, initial_goal: str): self.session_id session_id # 唯一标识用于重建 self.agent agent # 关联的 Agent 蓝图 self.initial_goal initial_goal self.state SessionState.CREATED # 状态CREATED, RUNNING, PAUSED, FINISHED, ERROR self.turns: List[TurnRecord] [] # 历史 Turn 记录 self.current_turn: Optional[TurnContext] None # 当前正在执行的 Turn 上下文 self.created_at: datetime self.updated_at: datetime # 用于超时判断 # 可能还包括对话历史、变量存储等SessionManager或SessionService负责 Session 的创建、获取、持久化和重建。class SessionManager: def __init__(self, storage_backend: SessionStorage): self.storage storage_backend # 持久化存储抽象 def create_session(self, agent: Agent, goal: str) - Session: session_id generate_uuid() session Session(session_idsession_id, agentagent, initial_goalgoal) self.storage.save(session) # 创建后立即持久化 return session def get_session(self, session_id: str) - Session: # 从存储中加载 Session 对象 return self.storage.load(session_id) def reconstruct_session(self, session_id: str) - Session: 核心的重建方法。 # 1. 从存储加载 Session 基础信息和所有历史 Turns session_data self.storage.load_full_state(session_id) # 2. 根据数据重新构建 Session 对象并恢复到中断前的状态 session self._hydrate_session(session_data) # 3. 可能还需要重新初始化一些运行时依赖如工具调用句柄 session self._reinitialize_runtime(session) # 4. 将状态设置为可继续运行如从 PAUSED 或 ERROR 改为 RUNNING session.state SessionState.RUNNING self.storage.save(session) # 更新状态 return sessionSessionStorage是一个抽象接口定义了save、load、load_full_state等方法。其实现可能是RedisSessionStorage、DatabaseSessionStorage或FileSessionStorage。持久化的数据必须包含足够的信息来完全重建 Session 和其历史。3.3 Turn 的执行流程Turn 是业务逻辑的轮次。我们可以在src/harness/turn/找到其执行器。核心类分析TurnExecutor驱动一个 Turn 的完整执行。class TurnExecutor: def __init__(self, llm_client, tool_executor): self.llm_client llm_client self.tool_executor tool_executor async def execute(self, session: Session, user_input: Optional[str] None) - TurnResult: 执行一个新的 Turn。 # 1. 创建 Turn 上下文并关联到 Session turn_context TurnContext(sessionsession, turn_idgenerate_turn_id()) session.current_turn turn_context # 2. 准备本次 Turn 的输入结合历史记忆和当前输入 prompt self._prepare_prompt(session, user_input) # 3. 进入 Step 执行循环 step_runner StepRunner(self.llm_client, self.tool_executor) while not turn_context.is_finished: # 决定下一个 Step 是什么思考、行动、观察、结束 next_step_type self._determine_next_step(turn_context) # 执行该 Step step_result await step_runner.run(next_step_type, turn_context) # 更新 Turn 上下文状态 turn_context.update_with_step_result(step_result) # 持久化中间状态可选但对于容错很重要 session.manager.storage.save_step(turn_context.turn_id, step_result) # 4. Turn 结束整理结果保存到 Session 历史 turn_record turn_context.to_record() session.turns.append(turn_record) session.current_turn None session.updated_at datetime.now() # 持久化整个 Session 的更新 session.manager.storage.save(session) return TurnResult.from_context(turn_context)关键点TurnExecutor的核心是一个由StepRunner驱动的循环。它根据当前上下文决定下一个 Step 的类型思考、行动等并执行它直到 Turn 完成例如Agent 给出了最终答案或决定不再需要工具。3.4 Step 的细化执行Step 是框架执行的最小单元。代码在src/harness/step/。核心类分析Step枚举或基类定义不同类型的 Step。class StepType(Enum): THINKING thinking # Agent 内部推理 ACTION action # 调用工具 OBSERVATION observation # 处理工具返回 FINAL final # 生成最终回复StepRunner根据 Step 类型执行具体的逻辑。class StepRunner: async def run(self, step_type: StepType, context: TurnContext) - StepResult: if step_type StepType.THINKING: return await self._run_thinking_step(context) elif step_type StepType.ACTION: return await self._run_action_step(context) elif step_type StepType.OBSERVATION: return await self._run_observation_step(context) # ... 其他类型 async def _run_thinking_step(self, context: TurnContext) - StepResult: 调用 LLM 进行链式思考ReAct, CoT 等模式。 messages self._build_messages_for_thinking(context) # 调用 LLM要求其输出思考过程和可能的行动决定 llm_response await self.llm_client.chat_completion(messages) # 解析 LLM 响应提取思考内容和下一步意图如调用哪个工具 parsed_thought self._parse_llm_response(llm_response) return StepResult( typeStepType.THINKING, contentparsed_thought.thought, metadata{next_action: parsed_thought.next_action} # 指示下一步是 ACTION ) async def _run_action_step(self, context: TurnContext) - StepResult: 执行工具调用。 # 从上一步THINKING的 metadata 中获取要执行的动作 action_to_take context.get_last_step().metadata[next_action] tool_name action_to_take.tool_name tool_args action_to_take.arguments # 通过 ToolExecutor 实际调用工具 tool_result await self.tool_executor.execute(tool_name, tool_args) return StepResult( typeStepType.ACTION, contentf调用工具 {tool_name} 完成。, metadata{tool_result: tool_result} # 工具执行结果供 OBSERVATION 使用 ) async def _run_observation_step(self, context: TurnContext) - StepResult: 处理工具执行结果并将其格式化为 Agent 可理解的观察。 tool_result context.get_last_step().metadata[tool_result] observation_text self._format_tool_result(tool_result) return StepResult( typeStepType.OBSERVATION, contentobservation_text, metadata{} # 观察完成后通常触发新一轮 THINKING )关键点StepRunner是连接框架逻辑与底层 LLM、工具调用的桥梁。每个 Step 的执行结果StepResult都会更新TurnContext从而影响后续 Step 的流转。3.5 执行引擎的协调作用Engine是顶层的协调者它对外提供简单的run或chat接口内部则串联起上述所有组件。核心类分析class HarnessEngine: def __init__(self, session_manager: SessionManager, turn_executor: TurnExecutor): self.session_manager session_manager self.turn_executor turn_executor async def run_agent(self, agent: Agent, goal: str, session_id: Optional[str] None) - Session: 启动一个 Agent 执行任务。 # 如果提供了 session_id尝试重建会话否则创建新会话。 if session_id: session self.session_manager.reconstruct_session(session_id) print(fSession {session_id} 已重建继续执行...) else: session self.session_manager.create_session(agent, goal) print(f新 Session {session.session_id} 已创建目标: {goal}) # 进入主循环直到任务完成或达到最大轮次 while session.state SessionState.RUNNING and len(session.turns) session.agent.config.max_turns: # 这里可以获取外部输入例如从消息队列对于自动任务输入可能为空或来自上一个Turn的结果 user_input await self._get_next_input(session) # 执行一个 Turn turn_result await self.turn_executor.execute(session, user_input) # 检查 Turn 结果是否意味着任务完成 if self._is_goal_achieved(turn_result, session.initial_goal): session.state SessionState.FINISHED self.session_manager.storage.save(session) break return session关键点Engine是入口。它处理了 Session 的创建/重建逻辑并管理着最高层次的任务循环Turn 循环。TurnExecutor管理 Step 循环StepRunner执行单个 Step。4. Session 重建的完整流程与实战模拟理解了各个组件后我们来模拟一个 Session 因服务重启而中断随后成功重建并继续执行的完整场景。这是生产环境稳定性的关键。4.1 场景设定假设我们有一个“天气查询助手”Agent其任务是“告诉我北京和上海明天的天气对比”。这个任务需要两个 TurnTurn 1: 查询北京天气。Turn 2: 查询上海天气然后进行对比总结。当服务在执行完Turn 1刚刚开始Turn 2的第一个THINKINGStep 时服务器意外重启。4.2 持久化存储的数据结构为了重建存储中必须保存足够详细的状态。我们来看一个简化的存储记录Session 表记录 (session_id:sess_123){ session_id: sess_123, agent_config: {name: weather_agent, model: deepseek-chat, ...}, initial_goal: 告诉我北京和上海明天的天气对比, state: RUNNING, current_turn_id: turn_456, // 中断时正在执行的 Turn ID created_at: 2024-05-27T10:00:00Z, updated_at: 2024-05-27T10:00:30Z // 中断前最后一次更新时间 }Turn 历史记录 (关联 session_id:sess_123)[ { turn_id: turn_123, sequence: 1, input: 告诉我北京和上海明天的天气对比, steps: [ {type: THINKING, content: 用户想对比北京和上海的天气。我需要先获取北京明天的天气。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 北京, date: tomorrow}}}}, {type: ACTION, content: 调用工具 get_weather 完成。, metadata: {tool_result: {city: 北京, weather: 晴, temp: 22°C}}}, {type: OBSERVATION, content: 北京明天天气晴气温22°C。, metadata: {}}, {type: THINKING, content: 已经拿到北京天气。接下来需要获取上海明天的天气然后对比。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 上海, date: tomorrow}}}} // 注意这个 THINKING Step 完成后本应进入 ACTION Step但服务中断了。 ], is_completed: false // Turn 1 实际上未完成因为思考后还没执行行动就中断了这里设计上可能有争议。更合理的可能是 Turn 1 已完成Turn 2 刚开始。 } ]更合理的设计每个 Turn 应在一个完整的“思考-行动-观察”循环后结束。所以 Turn 1 应在获得北京天气观察后结束。中断发生在 Turn 2 开始时。我们调整一下Turn 1 记录 (已完成){ turn_id: turn_123, sequence: 1, input: 告诉我北京和上海明天的天气对比, steps: [ {type: THINKING, content: 用户想对比北京和上海的天气。我需要先获取北京明天的天气。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 北京, date: tomorrow}}}}, {type: ACTION, content: 调用工具 get_weather 完成。, metadata: {tool_result: {city: 北京, weather: 晴, temp: 22°C}}}, {type: OBSERVATION, content: 北京明天天气晴气温22°C。, metadata: {}} ], final_output: 已查询到北京明天天气晴气温22°C。接下来将查询上海天气。, is_completed: true }Turn 2 记录 (进行中中断){ turn_id: turn_456, sequence: 2, input: null, // 续接上一轮可能没有新输入 steps: [ // 服务中断时Turn 2 只执行了一个 THINKING Step {type: THINKING, content: 现在需要查询上海明天的天气。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 上海, date: tomorrow}}}} ], is_completed: false }4.3 重建流程的代码模拟当服务重启收到继续执行sess_123的请求时SessionManager.reconstruct_session开始工作。# 在 SessionManager 中 async def reconstruct_session(self, session_id: str) - Session: # 1. 从存储加载完整数据 session_data await self.storage.load_session_data(session_id) turns_data await self.storage.load_turns_for_session(session_id) # 2. 重建 Agent 对象从配置 agent_config AgentConfig(**session_data[agent_config]) agent self.agent_builder.build(agent_config) # 使用 builder 重新构建 # 3. 重建 Session 对象 session Session( session_idsession_data[session_id], agentagent, initial_goalsession_data[initial_goal] ) session.state SessionState(session_data[state]) session.created_at session_data[created_at] session.updated_at session_data[updated_at] # 4. 重建历史 Turns for turn_data in turns_data: if turn_data[is_completed]: # 重建已完成的 Turn作为历史记录 turn_record TurnRecord.from_dict(turn_data) session.turns.append(turn_record) else: # 重建未完成的 Turn作为 current_turn # 这是关键恢复中断时的执行上下文 turn_context TurnContext.from_dict(turn_data, session) # 特别注意需要恢复 TurnContext 内部的状态机例如“下一个待执行的 Step 是什么” # 根据最后一个 Step 的 metadata[next_action]可以推断出下一步应是 ACTION last_step turn_context.steps[-1] if last_step.type StepType.THINKING: turn_context.next_step_type StepType.ACTION turn_context.pending_action last_step.metadata.get(next_action) session.current_turn turn_context # 5. 重新注入运行时依赖如工具执行器、LLM客户端它们可能在 Agent 中 # 这一步确保重建的 Session 可以继续调用工具和模型。 session.agent.llm_client self.llm_client session.agent.tool_registry self.tool_registry # 6. 保存重建后的状态可选标记为恢复中 await self.storage.save(session) return session4.4 继续执行Session 重建后被交还给HarnessEngine。引擎发现session.current_turn不为空即有一个中断的 Turn它会直接继续执行这个未完成的 Turn而不是开始一个新的 Turn。# 在 TurnExecutor.execute 中增加对恢复状态的处理 async def execute(self, session: Session, user_input: Optional[str] None) - TurnResult: # 判断是新的 Turn 还是恢复的 Turn if session.current_turn is not None: turn_context session.current_turn print(f恢复中断的 Turn: {turn_context.turn_id}) # 不需要创建新的 TurnContext else: # ... 创建新 TurnContext 的逻辑 ... # 接下来的 Step 循环会从 turn_context.next_step_type 开始 # 在我们的例子中next_step_type 是 ACTION step_runner StepRunner(self.llm_client, self.tool_executor) while not turn_context.is_finished: # 这里会取出 ACTION 作为第一个要执行的 Step next_step_type turn_context.next_step_type or self._determine_next_step(turn_context) step_result await step_runner.run(next_step_type, turn_context) # ... 更新上下文 ...于是系统会执行ACTIONStep调用get_weather工具查询上海天气然后自动进入OBSERVATIONStep接着可能再进入THINKINGStep 进行对比总结最终完成 Turn 2 和整个 Session。5. 常见问题与排查思路在实现或使用此类 Agent 框架时你可能会遇到以下问题问题现象可能原因排查思路与解决方案Session 重建后Agent 失忆了1. 持久化存储未包含完整的对话历史或记忆。2. 重建时Memory对象未正确恢复。3. 存储序列化/反序列化过程中丢失了数据。1. 检查存储的Session和Turn数据是否包含所有steps的content。2. 确保Memory类也实现了序列化接口并将其状态存入 Session。3. 使用 JSON 序列化时注意处理复杂的 Python 对象如 datetime可能需要自定义编码器。重建后执行逻辑错乱例如重复执行已完成的 Step。1.TurnContext的状态机如next_step_type、is_finished未正确持久化和恢复。2. 中断点判断逻辑有误。1. 在持久化TurnContext时确保保存其所有内部状态变量。2. 在reconstruct_session中仔细检查如何根据最后一个Step的类型和元数据来推算下一步该做什么。编写单元测试模拟各种中断场景。工具调用在重建后失败1. 工具依赖的外部连接如数据库、API 客户端在重建的 Session 中未重新初始化。2. 工具执行所需的临时状态丢失。1. 在SessionManager.reconstruct_session中确保重新注入所有运行时依赖ToolExecutor。2. 避免在工具实现中依赖不可序列化的对象或全局状态。工具应设计为无状态的。性能瓶颈持久化操作太频繁每个 Step 都进行全量 Session 存储I/O 压力大。1.异步持久化将保存操作放入后台队列不阻塞主执行流程。2.增量快照并非每个 Step 都全量保存可以定期或在关键节点如 Turn 结束时保存完整状态中间只存日志。3.使用更快的存储如 Redis。分布式环境下Session 被多个实例处理导致状态冲突两个服务器实例同时加载并修改同一个 Session。1.分布式锁在加载和保存 Session 时使用基于session_id的分布式锁如 Redis Lock。2.乐观锁在存储的数据中增加版本号version或updated_at保存时检查版本如果过期则重试或报错。6. 最佳实践与工程建议基于对 DeepSeek Harness 源码的分析我们可以提炼出一些设计 Agent 系统的通用最佳实践1. 状态设计明确区分静态配置与运行时状态静态配置Agent 的定义模型、系统提示词、工具列表、参数应可序列化且变化不频繁。重建时从配置重新构建。运行时状态Session、Turn、Step 历史、变量存储、记忆内容。这些是必须持久化的核心。黄金法则任何影响下一次 LLM 调用或工具调用的信息都必须成为状态的一部分并被持久化。2. 持久化策略权衡可靠性与性能存储选型根据 SLA 要求选择。Redis 性能好适合会话存储数据库PostgreSQL, MongoDB更可靠适合归档和复杂查询。序列化格式优先使用 JSON 或 MessagePack 等语言无关格式便于调试和多语言客户端。使用 Pydantic 等库可以简化验证和序列化。保存粒度关键点保存在 Turn 结束时必须保存。Step 级保存对于长任务每个 Step 后保存可减少重试成本但需考虑性能。可设置为可配置选项。异步保存务必使用异步操作避免阻塞 Agent 执行线程。3. 容错与幂等性工具调用的幂等性设计工具时尽量让它们支持幂等调用相同参数产生相同效果。这样在重试或重建后重复执行也不会造成问题如重复下单。LLM 调用的处理LLM 调用非幂等。重建后应避免向 LLM 发送完全相同的历史消息除非特意要求。通常将历史作为上下文重新发送是标准做法。设置超时与重试对 LLM 调用和工具调用设置合理的超时和重试机制并将这些尝试记录在状态中。4. 监控与可观测性结构化日志为每个 Session、Turn、Step 记录带有唯一 ID 的结构化日志。这比打印文本日志更利于追踪和聚合。指标收集收集关键指标如 Session 持续时间、Turn 数量、Step 数量、工具调用成功率、LLM 响应延迟等。追踪Tracing集成 OpenTelemetry 等工具可视化一个请求穿越 Agent、Turn、Step、LLM、工具的完整路径对于调试复杂问题至关重要。5. 测试策略单元测试针对StepRunner、TurnExecutor、SessionManager.reconstruct_session等核心单元进行测试。集成测试模拟完整的 Agent 执行流程并模拟服务中断验证 Session 重建后能正确继续。混沌测试在测试环境中随机终止服务进程验证系统的恢复能力。通过剖析 DeepSeek Harness 的源码我们不仅学习了一个优秀 Agent 框架的实现更重要的是掌握了一套构建可靠、可维护的智能代理系统的设计模式。从清晰的 Agent-Turn-Step 分层到细致的 Session 状态管理与重建机制这些思想都可以应用到我们自己的项目中去。下次当你设计一个需要处理长流程、有状态的任务系统时不妨回想一下 Harness 的这套架构它为解决状态持久化和故障恢复这个经典问题提供了一个优雅的范本。 从DeepSeek Harness源码解析AI Agent运行机制与Session重建
最近在尝试构建自己的 AI Agent 应用时发现很多框架要么过于复杂要么对运行过程的控制不够透明。直到深入研究了 DeepSeek Harness 的源码才真正理解了 Agent 从启动、执行到状态管理的完整生命周期。本文将带你从源码层面拆解 DeepSeek Harness 的核心运行机制特别是 Agent、Turn、Step 的协作关系以及 Session 重建这一关键容错策略。无论你是想深入理解 Agent 框架设计还是希望在自己的项目中实现类似的能力这篇文章都能提供一套完整的代码级分析思路。1. 背景与核心概念在深入源码之前我们有必要先厘清几个核心概念这有助于我们理解 DeepSeek Harness 的设计哲学和要解决的根本问题。1.1 什么是 DeepSeek HarnessDeepSeek Harness 是一个用于构建、管理和执行 AI Agent 的框架。你可以把它想象成一个“马具”Harness它的核心作用是将强大的大语言模型如 DeepSeek 系列模型“套”起来通过一套标准化的流程和控制机制让模型能够稳定、可靠地完成复杂的、多步骤的任务。它不是一个具体的 Agent而是一个生产 Agent 的“工厂”和“调度中心”。1.2 Agent、Turn、Step 与 Session 的关系这是理解整个框架运行机制的关键。我们可以用一个“对话式任务执行”的类比来理解Agent代理这是任务执行的主体。它被赋予一个目标Goal并拥有工具Tools、记忆Memory等能力。Agent 本身是一个相对静态的“配置实体”它定义了“谁”来执行任务以及“用什么”来执行。Session会话这是 Agent一次完整的任务执行过程。当你启动一个 Agent 去完成一个目标时就创建了一个 Session。Session 封装了这次任务执行的全部上下文包括初始目标、与环境的交互历史、当前状态等。它是动态的、有生命周期的。Turn轮次这是 Session 内部的一个宏观执行单元。通常Agent 完成一个目标需要多次“思考-行动-观察”的循环。每一次完整的循环可以看作一个 Turn。例如Agent 先“思考”需要搜索天气然后“执行”搜索工具最后“观察”搜索结果这构成一个 Turn。Turn 是面向业务逻辑的。Step步骤这是 Turn 内部更细粒度的执行步骤是框架与底层模型交互的最小单元。一个 Turn 通常由多个 Step 组成例如1) 生成推理链的 Step2) 调用工具的 Step3) 处理工具返回结果的 Step。Step 是面向框架执行引擎的。它们的关系可以概括为一个Agent启动后会创建一个Session来追踪本次任务。在该 Session 中Agent 通过完成多个Turn来逐步推进任务。而每一个 Turn则由框架拆解为一系列有序的Step来具体执行。1.3 为什么需要 Session 重建这是分布式、长耗时 Agent 系统中的一个经典问题。想象一下你的 Agent 任务可能需要运行几分钟甚至几小时在这期间服务可能因为部署、故障或负载均衡而重启。如果 Session 状态完全存在于内存中那么重启将导致任务丢失一切从头开始。Session 重建Session Recovery/Reconstruction就是为了解决这个问题。其核心思想是将 Session 的运行状态包括历史 Turns、Steps、工具调用结果、中间思考过程等持久化到外部存储如数据库、Redis。当服务中断后重启框架能够根据一个唯一的 Session ID从持久化存储中加载出完整的历史状态并让 Agent 从“中断点”继续执行而不是重新开始。这为生产环境提供了至关重要的可靠性和容错性。2. 环境准备与源码定位为了进行源码解析我们需要先准备好环境。本文的分析基于 DeepSeek Harness 的公开源码假设版本为v0.1.x左右具体 commit 可能变化但核心架构相对稳定。1. 获取源码git clone https://github.com/deepseek-ai/deepseek-harness.git cd deepseek-harness建议查看README.md和CONTRIBUTING.md了解项目结构和基本要求。2. 主要目录结构分析重点deepseek-harness/ ├── src/ │ ├── harness/ # 框架核心 │ │ ├── agent/ # Agent 定义、配置、构建器 │ │ ├── session/ # Session 生命周期管理、状态存储、重建逻辑 │ │ ├── turn/ # Turn 的执行流程控制 │ │ ├── step/ # Step 的定义与执行器 │ │ ├── engine/ # 执行引擎协调 Agent/Session/Turn/Step │ │ ├── memory/ # 记忆模块可能与会话状态结合 │ │ ├── tools/ # 工具系统 │ │ └── llm/ # 大语言模型集成层 │ └── common/ # 通用工具、常量、异常 ├── examples/ # 示例代码 ├── tests/ # 单元测试和集成测试 └── requirements.txt # Python 依赖3. 推荐的分析工具IDE使用 VS Code 或 PyCharm它们能提供优秀的代码导航和跳转。搜索在 IDE 中全局搜索关键词如SessionRecovery、reconstruct_session、persist_state、TurnExecutor、StepRunner。调试虽然不运行但可以查看examples/目录下的示例理解 API 的调用方式。版本说明Agent 框架发展迅速本文聚焦于核心设计模式和运行机制这些思想具有普适性。具体类名、方法签名可能随版本迭代而变化但分析思路是相通的。阅读时请关注设计思想而非绝对路径。3. 核心运行机制源码拆解现在我们深入到源码内部看看 Agent 是如何一步步动起来的。3.1 Agent 的初始化与配置Agent 的创建通常通过一个构建器Builder或配置类来完成。我们可以在src/harness/agent/目录下找到相关代码。核心类分析AgentConfig这是一个数据类可能使用 Pydantic定义了 Agent 的静态属性。# 示例性代码反映核心字段 class AgentConfig: name: str model: str # 使用的 LLM如 “deepseek-chat” system_prompt: str # 系统指令定义 Agent 角色和能力 tools: List[ToolDefinition] # 可用的工具列表 max_turns: Optional[int] 10 # 最大 Turn 数防止无限循环 memory_config: Optional[MemoryConfig] None # 记忆配置AgentBuilder或AgentFactory负责根据AgentConfig组装出可运行的Agent实例。这个过程会初始化 LLM 客户端、注册工具、设置记忆系统等。class AgentBuilder: def __init__(self, config: AgentConfig): self.config config self.llm_client self._init_llm_client() self.tool_registry self._init_tool_registry() def build(self) - Agent: 构建一个 Agent 实例。 return Agent( configself.config, llm_clientself.llm_client, tool_registryself.tool_registry, # ... 其他依赖注入 )关键点Agent对象本身通常不包含运行时状态那是 Session 的职责它更像是一个“蓝图”或“模板”。3.2 Session 的生命周期管理Session 是运行时状态的核心容器。相关代码主要在src/harness/session/。核心类分析Session类这是最重要的类之一。它持有一次任务执行的所有上下文。class Session: def __init__(self, session_id: str, agent: Agent, initial_goal: str): self.session_id session_id # 唯一标识用于重建 self.agent agent # 关联的 Agent 蓝图 self.initial_goal initial_goal self.state SessionState.CREATED # 状态CREATED, RUNNING, PAUSED, FINISHED, ERROR self.turns: List[TurnRecord] [] # 历史 Turn 记录 self.current_turn: Optional[TurnContext] None # 当前正在执行的 Turn 上下文 self.created_at: datetime self.updated_at: datetime # 用于超时判断 # 可能还包括对话历史、变量存储等SessionManager或SessionService负责 Session 的创建、获取、持久化和重建。class SessionManager: def __init__(self, storage_backend: SessionStorage): self.storage storage_backend # 持久化存储抽象 def create_session(self, agent: Agent, goal: str) - Session: session_id generate_uuid() session Session(session_idsession_id, agentagent, initial_goalgoal) self.storage.save(session) # 创建后立即持久化 return session def get_session(self, session_id: str) - Session: # 从存储中加载 Session 对象 return self.storage.load(session_id) def reconstruct_session(self, session_id: str) - Session: 核心的重建方法。 # 1. 从存储加载 Session 基础信息和所有历史 Turns session_data self.storage.load_full_state(session_id) # 2. 根据数据重新构建 Session 对象并恢复到中断前的状态 session self._hydrate_session(session_data) # 3. 可能还需要重新初始化一些运行时依赖如工具调用句柄 session self._reinitialize_runtime(session) # 4. 将状态设置为可继续运行如从 PAUSED 或 ERROR 改为 RUNNING session.state SessionState.RUNNING self.storage.save(session) # 更新状态 return sessionSessionStorage是一个抽象接口定义了save、load、load_full_state等方法。其实现可能是RedisSessionStorage、DatabaseSessionStorage或FileSessionStorage。持久化的数据必须包含足够的信息来完全重建 Session 和其历史。3.3 Turn 的执行流程Turn 是业务逻辑的轮次。我们可以在src/harness/turn/找到其执行器。核心类分析TurnExecutor驱动一个 Turn 的完整执行。class TurnExecutor: def __init__(self, llm_client, tool_executor): self.llm_client llm_client self.tool_executor tool_executor async def execute(self, session: Session, user_input: Optional[str] None) - TurnResult: 执行一个新的 Turn。 # 1. 创建 Turn 上下文并关联到 Session turn_context TurnContext(sessionsession, turn_idgenerate_turn_id()) session.current_turn turn_context # 2. 准备本次 Turn 的输入结合历史记忆和当前输入 prompt self._prepare_prompt(session, user_input) # 3. 进入 Step 执行循环 step_runner StepRunner(self.llm_client, self.tool_executor) while not turn_context.is_finished: # 决定下一个 Step 是什么思考、行动、观察、结束 next_step_type self._determine_next_step(turn_context) # 执行该 Step step_result await step_runner.run(next_step_type, turn_context) # 更新 Turn 上下文状态 turn_context.update_with_step_result(step_result) # 持久化中间状态可选但对于容错很重要 session.manager.storage.save_step(turn_context.turn_id, step_result) # 4. Turn 结束整理结果保存到 Session 历史 turn_record turn_context.to_record() session.turns.append(turn_record) session.current_turn None session.updated_at datetime.now() # 持久化整个 Session 的更新 session.manager.storage.save(session) return TurnResult.from_context(turn_context)关键点TurnExecutor的核心是一个由StepRunner驱动的循环。它根据当前上下文决定下一个 Step 的类型思考、行动等并执行它直到 Turn 完成例如Agent 给出了最终答案或决定不再需要工具。3.4 Step 的细化执行Step 是框架执行的最小单元。代码在src/harness/step/。核心类分析Step枚举或基类定义不同类型的 Step。class StepType(Enum): THINKING thinking # Agent 内部推理 ACTION action # 调用工具 OBSERVATION observation # 处理工具返回 FINAL final # 生成最终回复StepRunner根据 Step 类型执行具体的逻辑。class StepRunner: async def run(self, step_type: StepType, context: TurnContext) - StepResult: if step_type StepType.THINKING: return await self._run_thinking_step(context) elif step_type StepType.ACTION: return await self._run_action_step(context) elif step_type StepType.OBSERVATION: return await self._run_observation_step(context) # ... 其他类型 async def _run_thinking_step(self, context: TurnContext) - StepResult: 调用 LLM 进行链式思考ReAct, CoT 等模式。 messages self._build_messages_for_thinking(context) # 调用 LLM要求其输出思考过程和可能的行动决定 llm_response await self.llm_client.chat_completion(messages) # 解析 LLM 响应提取思考内容和下一步意图如调用哪个工具 parsed_thought self._parse_llm_response(llm_response) return StepResult( typeStepType.THINKING, contentparsed_thought.thought, metadata{next_action: parsed_thought.next_action} # 指示下一步是 ACTION ) async def _run_action_step(self, context: TurnContext) - StepResult: 执行工具调用。 # 从上一步THINKING的 metadata 中获取要执行的动作 action_to_take context.get_last_step().metadata[next_action] tool_name action_to_take.tool_name tool_args action_to_take.arguments # 通过 ToolExecutor 实际调用工具 tool_result await self.tool_executor.execute(tool_name, tool_args) return StepResult( typeStepType.ACTION, contentf调用工具 {tool_name} 完成。, metadata{tool_result: tool_result} # 工具执行结果供 OBSERVATION 使用 ) async def _run_observation_step(self, context: TurnContext) - StepResult: 处理工具执行结果并将其格式化为 Agent 可理解的观察。 tool_result context.get_last_step().metadata[tool_result] observation_text self._format_tool_result(tool_result) return StepResult( typeStepType.OBSERVATION, contentobservation_text, metadata{} # 观察完成后通常触发新一轮 THINKING )关键点StepRunner是连接框架逻辑与底层 LLM、工具调用的桥梁。每个 Step 的执行结果StepResult都会更新TurnContext从而影响后续 Step 的流转。3.5 执行引擎的协调作用Engine是顶层的协调者它对外提供简单的run或chat接口内部则串联起上述所有组件。核心类分析class HarnessEngine: def __init__(self, session_manager: SessionManager, turn_executor: TurnExecutor): self.session_manager session_manager self.turn_executor turn_executor async def run_agent(self, agent: Agent, goal: str, session_id: Optional[str] None) - Session: 启动一个 Agent 执行任务。 # 如果提供了 session_id尝试重建会话否则创建新会话。 if session_id: session self.session_manager.reconstruct_session(session_id) print(fSession {session_id} 已重建继续执行...) else: session self.session_manager.create_session(agent, goal) print(f新 Session {session.session_id} 已创建目标: {goal}) # 进入主循环直到任务完成或达到最大轮次 while session.state SessionState.RUNNING and len(session.turns) session.agent.config.max_turns: # 这里可以获取外部输入例如从消息队列对于自动任务输入可能为空或来自上一个Turn的结果 user_input await self._get_next_input(session) # 执行一个 Turn turn_result await self.turn_executor.execute(session, user_input) # 检查 Turn 结果是否意味着任务完成 if self._is_goal_achieved(turn_result, session.initial_goal): session.state SessionState.FINISHED self.session_manager.storage.save(session) break return session关键点Engine是入口。它处理了 Session 的创建/重建逻辑并管理着最高层次的任务循环Turn 循环。TurnExecutor管理 Step 循环StepRunner执行单个 Step。4. Session 重建的完整流程与实战模拟理解了各个组件后我们来模拟一个 Session 因服务重启而中断随后成功重建并继续执行的完整场景。这是生产环境稳定性的关键。4.1 场景设定假设我们有一个“天气查询助手”Agent其任务是“告诉我北京和上海明天的天气对比”。这个任务需要两个 TurnTurn 1: 查询北京天气。Turn 2: 查询上海天气然后进行对比总结。当服务在执行完Turn 1刚刚开始Turn 2的第一个THINKINGStep 时服务器意外重启。4.2 持久化存储的数据结构为了重建存储中必须保存足够详细的状态。我们来看一个简化的存储记录Session 表记录 (session_id:sess_123){ session_id: sess_123, agent_config: {name: weather_agent, model: deepseek-chat, ...}, initial_goal: 告诉我北京和上海明天的天气对比, state: RUNNING, current_turn_id: turn_456, // 中断时正在执行的 Turn ID created_at: 2024-05-27T10:00:00Z, updated_at: 2024-05-27T10:00:30Z // 中断前最后一次更新时间 }Turn 历史记录 (关联 session_id:sess_123)[ { turn_id: turn_123, sequence: 1, input: 告诉我北京和上海明天的天气对比, steps: [ {type: THINKING, content: 用户想对比北京和上海的天气。我需要先获取北京明天的天气。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 北京, date: tomorrow}}}}, {type: ACTION, content: 调用工具 get_weather 完成。, metadata: {tool_result: {city: 北京, weather: 晴, temp: 22°C}}}, {type: OBSERVATION, content: 北京明天天气晴气温22°C。, metadata: {}}, {type: THINKING, content: 已经拿到北京天气。接下来需要获取上海明天的天气然后对比。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 上海, date: tomorrow}}}} // 注意这个 THINKING Step 完成后本应进入 ACTION Step但服务中断了。 ], is_completed: false // Turn 1 实际上未完成因为思考后还没执行行动就中断了这里设计上可能有争议。更合理的可能是 Turn 1 已完成Turn 2 刚开始。 } ]更合理的设计每个 Turn 应在一个完整的“思考-行动-观察”循环后结束。所以 Turn 1 应在获得北京天气观察后结束。中断发生在 Turn 2 开始时。我们调整一下Turn 1 记录 (已完成){ turn_id: turn_123, sequence: 1, input: 告诉我北京和上海明天的天气对比, steps: [ {type: THINKING, content: 用户想对比北京和上海的天气。我需要先获取北京明天的天气。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 北京, date: tomorrow}}}}, {type: ACTION, content: 调用工具 get_weather 完成。, metadata: {tool_result: {city: 北京, weather: 晴, temp: 22°C}}}, {type: OBSERVATION, content: 北京明天天气晴气温22°C。, metadata: {}} ], final_output: 已查询到北京明天天气晴气温22°C。接下来将查询上海天气。, is_completed: true }Turn 2 记录 (进行中中断){ turn_id: turn_456, sequence: 2, input: null, // 续接上一轮可能没有新输入 steps: [ // 服务中断时Turn 2 只执行了一个 THINKING Step {type: THINKING, content: 现在需要查询上海明天的天气。, metadata: {next_action: {tool_name: get_weather, arguments: {city: 上海, date: tomorrow}}}} ], is_completed: false }4.3 重建流程的代码模拟当服务重启收到继续执行sess_123的请求时SessionManager.reconstruct_session开始工作。# 在 SessionManager 中 async def reconstruct_session(self, session_id: str) - Session: # 1. 从存储加载完整数据 session_data await self.storage.load_session_data(session_id) turns_data await self.storage.load_turns_for_session(session_id) # 2. 重建 Agent 对象从配置 agent_config AgentConfig(**session_data[agent_config]) agent self.agent_builder.build(agent_config) # 使用 builder 重新构建 # 3. 重建 Session 对象 session Session( session_idsession_data[session_id], agentagent, initial_goalsession_data[initial_goal] ) session.state SessionState(session_data[state]) session.created_at session_data[created_at] session.updated_at session_data[updated_at] # 4. 重建历史 Turns for turn_data in turns_data: if turn_data[is_completed]: # 重建已完成的 Turn作为历史记录 turn_record TurnRecord.from_dict(turn_data) session.turns.append(turn_record) else: # 重建未完成的 Turn作为 current_turn # 这是关键恢复中断时的执行上下文 turn_context TurnContext.from_dict(turn_data, session) # 特别注意需要恢复 TurnContext 内部的状态机例如“下一个待执行的 Step 是什么” # 根据最后一个 Step 的 metadata[next_action]可以推断出下一步应是 ACTION last_step turn_context.steps[-1] if last_step.type StepType.THINKING: turn_context.next_step_type StepType.ACTION turn_context.pending_action last_step.metadata.get(next_action) session.current_turn turn_context # 5. 重新注入运行时依赖如工具执行器、LLM客户端它们可能在 Agent 中 # 这一步确保重建的 Session 可以继续调用工具和模型。 session.agent.llm_client self.llm_client session.agent.tool_registry self.tool_registry # 6. 保存重建后的状态可选标记为恢复中 await self.storage.save(session) return session4.4 继续执行Session 重建后被交还给HarnessEngine。引擎发现session.current_turn不为空即有一个中断的 Turn它会直接继续执行这个未完成的 Turn而不是开始一个新的 Turn。# 在 TurnExecutor.execute 中增加对恢复状态的处理 async def execute(self, session: Session, user_input: Optional[str] None) - TurnResult: # 判断是新的 Turn 还是恢复的 Turn if session.current_turn is not None: turn_context session.current_turn print(f恢复中断的 Turn: {turn_context.turn_id}) # 不需要创建新的 TurnContext else: # ... 创建新 TurnContext 的逻辑 ... # 接下来的 Step 循环会从 turn_context.next_step_type 开始 # 在我们的例子中next_step_type 是 ACTION step_runner StepRunner(self.llm_client, self.tool_executor) while not turn_context.is_finished: # 这里会取出 ACTION 作为第一个要执行的 Step next_step_type turn_context.next_step_type or self._determine_next_step(turn_context) step_result await step_runner.run(next_step_type, turn_context) # ... 更新上下文 ...于是系统会执行ACTIONStep调用get_weather工具查询上海天气然后自动进入OBSERVATIONStep接着可能再进入THINKINGStep 进行对比总结最终完成 Turn 2 和整个 Session。5. 常见问题与排查思路在实现或使用此类 Agent 框架时你可能会遇到以下问题问题现象可能原因排查思路与解决方案Session 重建后Agent 失忆了1. 持久化存储未包含完整的对话历史或记忆。2. 重建时Memory对象未正确恢复。3. 存储序列化/反序列化过程中丢失了数据。1. 检查存储的Session和Turn数据是否包含所有steps的content。2. 确保Memory类也实现了序列化接口并将其状态存入 Session。3. 使用 JSON 序列化时注意处理复杂的 Python 对象如 datetime可能需要自定义编码器。重建后执行逻辑错乱例如重复执行已完成的 Step。1.TurnContext的状态机如next_step_type、is_finished未正确持久化和恢复。2. 中断点判断逻辑有误。1. 在持久化TurnContext时确保保存其所有内部状态变量。2. 在reconstruct_session中仔细检查如何根据最后一个Step的类型和元数据来推算下一步该做什么。编写单元测试模拟各种中断场景。工具调用在重建后失败1. 工具依赖的外部连接如数据库、API 客户端在重建的 Session 中未重新初始化。2. 工具执行所需的临时状态丢失。1. 在SessionManager.reconstruct_session中确保重新注入所有运行时依赖ToolExecutor。2. 避免在工具实现中依赖不可序列化的对象或全局状态。工具应设计为无状态的。性能瓶颈持久化操作太频繁每个 Step 都进行全量 Session 存储I/O 压力大。1.异步持久化将保存操作放入后台队列不阻塞主执行流程。2.增量快照并非每个 Step 都全量保存可以定期或在关键节点如 Turn 结束时保存完整状态中间只存日志。3.使用更快的存储如 Redis。分布式环境下Session 被多个实例处理导致状态冲突两个服务器实例同时加载并修改同一个 Session。1.分布式锁在加载和保存 Session 时使用基于session_id的分布式锁如 Redis Lock。2.乐观锁在存储的数据中增加版本号version或updated_at保存时检查版本如果过期则重试或报错。6. 最佳实践与工程建议基于对 DeepSeek Harness 源码的分析我们可以提炼出一些设计 Agent 系统的通用最佳实践1. 状态设计明确区分静态配置与运行时状态静态配置Agent 的定义模型、系统提示词、工具列表、参数应可序列化且变化不频繁。重建时从配置重新构建。运行时状态Session、Turn、Step 历史、变量存储、记忆内容。这些是必须持久化的核心。黄金法则任何影响下一次 LLM 调用或工具调用的信息都必须成为状态的一部分并被持久化。2. 持久化策略权衡可靠性与性能存储选型根据 SLA 要求选择。Redis 性能好适合会话存储数据库PostgreSQL, MongoDB更可靠适合归档和复杂查询。序列化格式优先使用 JSON 或 MessagePack 等语言无关格式便于调试和多语言客户端。使用 Pydantic 等库可以简化验证和序列化。保存粒度关键点保存在 Turn 结束时必须保存。Step 级保存对于长任务每个 Step 后保存可减少重试成本但需考虑性能。可设置为可配置选项。异步保存务必使用异步操作避免阻塞 Agent 执行线程。3. 容错与幂等性工具调用的幂等性设计工具时尽量让它们支持幂等调用相同参数产生相同效果。这样在重试或重建后重复执行也不会造成问题如重复下单。LLM 调用的处理LLM 调用非幂等。重建后应避免向 LLM 发送完全相同的历史消息除非特意要求。通常将历史作为上下文重新发送是标准做法。设置超时与重试对 LLM 调用和工具调用设置合理的超时和重试机制并将这些尝试记录在状态中。4. 监控与可观测性结构化日志为每个 Session、Turn、Step 记录带有唯一 ID 的结构化日志。这比打印文本日志更利于追踪和聚合。指标收集收集关键指标如 Session 持续时间、Turn 数量、Step 数量、工具调用成功率、LLM 响应延迟等。追踪Tracing集成 OpenTelemetry 等工具可视化一个请求穿越 Agent、Turn、Step、LLM、工具的完整路径对于调试复杂问题至关重要。5. 测试策略单元测试针对StepRunner、TurnExecutor、SessionManager.reconstruct_session等核心单元进行测试。集成测试模拟完整的 Agent 执行流程并模拟服务中断验证 Session 重建后能正确继续。混沌测试在测试环境中随机终止服务进程验证系统的恢复能力。通过剖析 DeepSeek Harness 的源码我们不仅学习了一个优秀 Agent 框架的实现更重要的是掌握了一套构建可靠、可维护的智能代理系统的设计模式。从清晰的 Agent-Turn-Step 分层到细致的 Session 状态管理与重建机制这些思想都可以应用到我们自己的项目中去。下次当你设计一个需要处理长流程、有状态的任务系统时不妨回想一下 Harness 的这套架构它为解决状态持久化和故障恢复这个经典问题提供了一个优雅的范本。