1. 先搞清楚 DeepSeek Harness 到底解决了什么问题如果你正在研究或开发 AI Agent尤其是基于 DeepSeek 这类大语言模型构建的 Agent那么你很可能遇到过这些头疼的事Agent 的执行状态怎么管理一次对话里模型思考、工具调用、用户反馈这些来回交互怎么组织任务执行到一半中断了怎么让它接着跑而不是重头再来还有怎么把 Agent 的“记忆”和“思考过程”保存下来方便调试和复盘DeepSeek Harness 这个框架核心解决的就是Agent 执行流程的标准化、可观测和可持久化问题。它不是另一个告诉你“怎么调用 API”的简单封装而是深入到 Agent 每一次“呼吸”Turn和“动作”Step的底层提供了一套机制来记录、控制和恢复整个执行会话Session。最值得关注的点是它的三个核心概念Turn回合、Step步骤和 Session会话。很多 Agent 框架只关心“输入-输出”但 Harness 把一次复杂的 Agent 交互拆解成可管理的单元。一个Turn可能包含用户的一次提问而一个Turn内部又可能包含模型思考、调用工具、解析结果等多个Step。Session则把所有这些Turn和Step连同当时的上下文、工具执行结果、模型响应都打包保存起来。这意味着什么意味着你可以精确调试不再是看最终输出对不对而是能回溯到模型在某个Step里到底“想”了什么工具返回了什么。状态恢复Agent 执行到第 5 步因为网络问题挂了没关系从保存的Session里恢复直接从第 5 步接着跑不用浪费前面的 token 和计算。流程标准化无论是简单的问答还是复杂的多工具协作流程都通过Turn/Step来组织代码结构清晰不同 Agent 之间的行为也更容易对比和理解。所以这篇文章不是泛泛而谈 Agent 概念而是直接切入 DeepSeek Harness 的源码看它如何用Turn、Step、Session这套机制把 Agent 从“一锤子买卖”变成可管理、可调试、可恢复的持续服务。无论你是想学习成熟 Agent 框架的设计思想还是打算在自己的项目里引入类似的状态管理这里的解析都能给你直接的参考。2. 核心三要素Turn, Step, Session 的职责与关系在跑代码之前必须先把这三个核心对象的关系和各自职责理清楚。很多人在看源码时容易晕就是因为没搞明白它们的分层和协作逻辑。2.1 Session承载整个交互旅程的容器你可以把Session理解为一个项目文件夹。一次完整的、可能有来有回的 Agent 交互任务比如“帮我分析这份数据并生成报告”就是一个Session。它的核心职责是持久化保存整个交互过程的所有数据包括所有的Turn、Step以及初始的系统提示词System Prompt、会话元数据如创建时间、关联的模型配置等。状态管理维护当前会话的整体状态例如是否已完成、是否出错。上下文维护它不直接处理单次交互但持有产生这些交互的完整历史和环境。在源码中Session类通常包含以下关键属性具体字段名可能因版本略有差异但思想一致# 示例性结构非直接源码 class Session: id: str # 会话唯一ID created_at: datetime system_prompt: str # 本次会话的系统指令 turns: List[Turn] # 按顺序存放的所有回合 metadata: Dict # 自定义元数据如用户ID、任务类型 status: SessionStatus # 如 RUNNING, COMPLETED, ERRORSession是顶层容器它本身不执行逻辑而是通过管理Turn列表来记录发生了什么。2.2 Turn一次完整的“请求-响应”循环Turn是Session内部的主要交互单元。通常用户的一次输入一个 Prompt以及 Agent 为完成这个输入所进行的一系列工作构成一个Turn。例如用户问“北京今天的天气怎么样” - 这触发一个新的Turn。Agent 需要理解意图 - 调用天气查询工具 - 解析工具结果 - 组织自然语言回复。这一连串的动作都属于同一个Turn。所以Turn的核心职责是组织一次完整的 QA 循环它关联一个用户输入或上一个 Agent 输出并产出最终的 Agent 回复。管理 Step 序列它将完成此次回复所需的复杂过程分解为多个线性的Step。维护回合级上下文它保存了此回合初始的 Prompt以及最终生成的 Response。源码中的Turn可能长这样class Turn: id: str # 回合ID通常在Session内唯一 session_id: str # 归属哪个Session prompt: str # 用户或上游输入 steps: List[Step] # 按顺序执行的所有步骤 response: Optional[str] # 本回合的最终输出 created_at: datetime关键理解一个Turn的response并不是第一个Step的产出而是所有Step执行完毕后最终整合的结果。Turn是逻辑单元Step是执行单元。2.3 Step原子化的执行动作Step是Harness 执行引擎的最小工作单元。一个Turn内部Agent 的“思考-行动”过程被分解成一个个Step。常见的Step类型包括LLM Inference Step调用大模型生成文本可能是思考链也可能是直接回复或工具调用请求。Tool Call Step执行一个具体的工具比如调用 API、查询数据库、运行代码。Tool Result Step处理工具执行返回的结果并将其格式化为模型可以理解的文本。Parse Step解析模型输出例如从文本中提取出结构化的工具调用参数。每个Step都应该是原子的、结果明确的。它的职责非常清晰执行单一逻辑只做一件事并产生一个明确的结果。记录输入输出保存执行前的状态输入和执行后的状态输出。这是后期调试和恢复的基石。维护执行状态记录自己是成功SUCCESS、失败ERROR还是等待中PENDING。源码中Step的结构示例class Step: id: str # 步骤ID turn_id: str # 归属哪个Turn type: StepType # 如 “llm”, “tool_call”, “tool_result” input: Dict # 执行所需的输入数据例如对于LLM Step就是messages output: Optional[Dict] # 执行后的输出数据例如对于LLM Step就是模型的response对象 status: StepStatus # PENDING, RUNNING, SUCCESS, ERROR error: Optional[str] # 如果失败错误信息 created_at: datetime updated_at: datetime2.4 三者的协作关系与数据流用一个简单的“查询天气”例子把三者串起来创建 Session用户开始一个新任务。系统创建一个SessionID 为sess_001并设置系统提示词“你是一个有帮助的助手可以使用工具。”创建 Turn 1用户输入“北京今天的天气如何” 系统在sess_001下创建一个新的TurnID 为turn_1prompt字段记录用户问题。执行 Step 1 (LLM Inference)Harness 引擎开始处理turn_1。它先创建一个Step类型为llminput是包含系统提示词和用户问题的 messages 列表。执行后output是模型的回复比如“我需要调用天气查询工具来获取信息。call_tool name“get_weather” city“北京”。” 此Step状态标记为SUCCESS。执行 Step 2 (Tool Call)引擎解析出工具调用指令创建新的Step类型为tool_callinput是{“name”: “get_weather”, “params”: {“city”: “北京”}}。执行工具调用真实天气 APIoutput是 API 返回的原始数据{“city”: “Beijing”, “temp”: 22, “condition”: “Sunny”}。状态标记为SUCCESS。执行 Step 3 (LLM Inference Again)引擎将工具结果格式化再次创建llm类型的Stepinput是包含工具结果的新 messages。模型生成最终回复“北京今天天气晴朗气温 22 摄氏度。” 此Step的output记录模型回复。完成 Turn所有Step执行完毕引擎将最终回复 “北京今天天气晴朗气温 22 摄氏度。” 赋值给turn_1.response。Session 更新turn_1被加入到sess_001.turns列表中。Session的状态可能更新。整个数据流是Session - Turn - Step (链式执行) - 更新 Turn - 更新 Session。Step的输入输出构成了一个可追溯的链条这正是 Harness 实现可观测和可恢复的基础。3. 从源码看 Agent 运行机制引擎如何驱动 Step理解了静态结构我们来看动态过程Harness 的引擎是如何让一个Session从一个Turn的Prompt开始一步步执行Step最终产生Response的。这是整个框架最核心的“发动机”部分。3.1 执行引擎的核心循环Harness 内部会有一个Executor或Engine类具体类名可能不同它负责协调整个执行流程。其核心逻辑是一个状态机驱动的循环大致如下# 概念性伪代码展示引擎逻辑 class HarnessEngine: def run_turn(self, session: Session, turn: Turn): # 1. 初始化将Turn状态设为RUNNING创建第一个Step通常是LLM Step turn.status TurnStatus.RUNNING current_step self._create_initial_step(turn) # 2. 核心执行循环 while not self._is_turn_complete(turn): # 2.1 执行当前Step step_output self._execute_step(current_step, session.context) current_step.output step_output current_step.status StepStatus.SUCCESS if success else StepStatus.ERROR # 2.2 判断Step执行结果决定下一步动作 if current_step.type StepType.LLM: # 分析模型输出看是否需要调用工具 if self._needs_tool_call(current_step.output): # 创建Tool Call Step next_step self._create_tool_call_step(turn, current_step.output) else: # 模型直接给出最终答案Turn可以结束了 turn.response self._extract_final_response(current_step.output) break elif current_step.type StepType.TOOL_CALL: # 工具调用完毕创建Tool Result Step来包装结果 next_step self._create_tool_result_step(turn, current_step.output) elif current_step.type StepType.TOOL_RESULT: # 工具结果已就绪需要再次让模型处理创建新的LLM Step next_step self._create_llm_step_with_result(turn, current_step.output) # 2.3 将新Step加入Turn并设为当前Step继续循环 turn.steps.append(next_step) current_step next_step # 3. 循环结束标记Turn完成 turn.status TurnStatus.COMPLETED session.turns.append(turn)这个循环的关键在于_execute_step方法和后续的“下一步决策”逻辑。_execute_step是一个分发器根据Step的类型调用不同的处理器Handler。3.2 各类型 Step 的执行处理器源码中会有像LLMStepHandler,ToolCallStepHandler,ToolResultStepHandler这样的类。它们的execute方法决定了具体干什么。LLMStepHandler.execute(step):输入从step.input中提取 messages历史对话 最新上下文。动作调用配置好的 DeepSeek或其他模型 API。输出将模型的原始响应可能包含思考过程、工具调用标记存入step.output。关键细节这里会处理模型的流式输出如果支持并可能解析function calling或类似格式。模型输出中的工具调用指令是触发后续ToolCallStep的关键信号。ToolCallStepHandler.execute(step):输入从step.input中提取工具名称和参数字典。动作在注册的工具库中查找对应工具传入参数并执行。输出将工具执行后的原始结果可能是任何 Python 对象存入step.output。关键细节工具执行必须是安全的、可捕获异常的。任何异常都应被捕获并转化为step.status ERROR和step.error信息而不是让整个引擎崩溃。ToolResultStepHandler.execute(step):输入通常是上一个ToolCallStep的output。动作将工具执行的原始结果如 JSON、文本、数字格式化成一段自然语言描述以便拼接到后续给模型的 messages 中。输出格式化后的文本描述存入step.output。关键细节这个格式化过程很重要。直接给模型扔一个 Python 对象或复杂的 JSON模型可能无法很好理解。通常格式化成“工具 XXX 返回的结果是……”。3.3 上下文Context的传递与组装你可能会问每个Step的input是怎么来的特别是LLMStep的 messages它需要包含之前所有相关的历史。这就是Session上下文Context的作用。引擎在创建新的LLMStep时会调用一个_build_messages_for_step的方法。这个方法会获取Session.system_prompt作为第一条 system message。遍历当前Turn之前的所有Step根据Step的类型和输出构建出完整的对话历史。对于LLMStep将其输出内容作为assistant消息。对于ToolCallStep和ToolResultStep通常将它们合并或单独作为tool或user消息取决于模型支持的格式如 OpenAI 的tool_calls角色。将最新的用户输入或工具结果作为最新的user消息加入。这样每次模型调用都能获得截至当前步骤的完整上下文确保对话的连贯性。Session和Turn/Step的层级结构使得这种上下文的组装变得清晰且高效。4. Session 重建如何让 Agent“接着上次的聊”Session重建是 Harness 框架非常强大的一个特性。它意味着你可以把一次未完成或已完成的 Agent 交互完整地保存下来比如存到数据库或文件然后在另一个时间、另一个进程甚至另一台机器上精确地恢复到保存时的状态继续执行或重新分析。这为以下场景提供了可能长时任务/持久化 Agent一个需要运行数小时甚至数天的自动化 Agent服务器需要重启时可以从断点恢复。调试与审计可以随时加载任意历史Session复现问题查看每一步的中间状态。用户会话恢复在聊天应用中用户关闭页面后再打开可以无缝继续之前的对话。4.1 Session 的序列化与持久化要实现重建首先要能保存。Harness 的Session、Turn、Step对象必须是可序列化的。通常这意味着它们的所有属性都是基本数据类型str, int, dict, list, None或 datetime或者实现了自定义的to_dict()和from_dict()方法。源码中通常会有一个SessionStore或Persistence抽象层定义保存和加载的接口class SessionStore: def save(self, session: Session) - str: # 将Session对象转换为字典然后存入数据库/文件 session_dict session.to_dict() # ... 存储逻辑返回session_id return session.id def load(self, session_id: str) - Session: # 从数据库/文件读取字典数据 session_dict ... # 读取逻辑 # 通过from_dict方法重建Session对象以及其下的Turn、Step对象 return Session.from_dict(session_dict)to_dict()方法需要递归地将所有嵌套的Turn和Step都转换成字典。from_dict()则是反向操作从字典数据重建出完整的对象树。这里的关键是保证对象引用的完整性比如每个Step的turn_id必须对应到正确的Turn对象。4.2 重建 Session 并继续执行假设我们有一个保存的SessionJSON 文件里面记录了一个未完成的天气查询任务刚执行完ToolCallStep拿到了天气数据但还没让模型生成最终回复。重建并继续执行的流程如下# 1. 从存储中加载Session session_store SessionStore() loaded_session session_store.load(“sess_001”) # 2. 定位到需要继续的Turn和Step # 通常我们会找状态为RUNNING或最后一个未完成的Turn last_turn loaded_session.turns[-1] if last_turn.status TurnStatus.RUNNING: # 3. 重建执行引擎并注入已加载的Session engine HarnessEngine(sessionloaded_session) # 4. 关键让引擎从特定状态“热启动” # 引擎需要知道从哪个Turn的哪个Step开始执行。 # 一种常见设计Turn的steps列表里最后一个Step的状态决定了下一步。 last_step last_turn.steps[-1] if last_step.status StepStatus.SUCCESS: # 上一步成功了需要决定下一步做什么引擎内部逻辑 # 例如上一步是ToolCallStep成功那么下一步应该是创建ToolResultStep next_step_type engine._decide_next_step_type(last_step) new_step Step(..., typenext_step_type, ...) last_turn.steps.append(new_step) # 引擎继续执行这个新创建的Step engine.run_step(new_step) elif last_step.status StepStatus.ERROR: # 上一步出错了可以尝试重试或进入错误处理流程 engine.handle_step_error(last_step) # ... 其他状态处理重建的精髓在于恢复完整的上下文。引擎拿到重建的Session对象后它看到的last_turn.steps列表和每个Step的input/output与中断前一模一样。因此当它执行_build_messages_for_step来准备下一次 LLM 调用时组装出的 messages 历史也是完全连续的模型感觉不到中断。4.3 重建过程中的挑战与源码应对在源码中重建功能会面临几个挑战Harness 通常这样处理外部资源依赖ToolCallStep中可能包含数据库连接、API 客户端等不可序列化的对象。解决方案是在to_dict()时只保存工具的配置信息如 API endpoint, credentials key在from_dict()重建时根据配置信息重新初始化这些客户端。这通常依赖一个全局的、可配置的ToolRegistry。模型状态LLM 本身是无状态的所以模型调用不需要特殊恢复。只需要恢复调用模型时的 messages 上下文即可。执行环境差异重建后的环境可能和保存时不同如 Python 包版本、工具依赖。Harness 框架本身无法完全解决但良好的Step设计输入输出明确可以让你快速定位是否是环境差异导致的重建后执行失败。5. 实战基于源码思想设计一个简易 Agent 状态管理系统看懂了原理最好的巩固方式就是动手设计一个简化版。我们不直接复制 Harness 源码而是借鉴其Turn/Step/Session的核心思想构建一个能管理 Agent 状态的小系统。这将帮助你真正理解数据如何流动。5.1 定义数据模型首先定义我们自己的简易数据类。from enum import Enum from datetime import datetime from typing import Dict, List, Optional, Any from pydantic import BaseModel # 使用Pydantic方便验证和序列化 class StepStatus(str, Enum): PENDING “pending” RUNNING “running” SUCCESS “success” ERROR “error” class StepType(str, Enum): LLM “llm” TOOL_CALL “tool_call” TOOL_RESULT “tool_result” class Step(BaseModel): id: str turn_id: str type: StepType status: StepStatus StepStatus.PENDING input: Dict[str, Any] # 执行输入 output: Optional[Dict[str, Any]] None # 执行输出 error: Optional[str] None created_at: datetime datetime.now() updated_at: datetime datetime.now() class Config: use_enum_values True # 序列化时使用枚举值 class TurnStatus(str, Enum): RUNNING “running” COMPLETED “completed” ERROR “error” class Turn(BaseModel): id: str session_id: str prompt: str # 用户输入 response: Optional[str] None # 最终回复 status: TurnStatus TurnStatus.RUNNING steps: List[Step] [] created_at: datetime datetime.now() class SessionStatus(str, Enum): ACTIVE “active” FINISHED “finished” class Session(BaseModel): id: str system_prompt: str “You are a helpful assistant.” status: SessionStatus SessionStatus.ACTIVE turns: List[Turn] [] metadata: Dict[str, Any] {} created_at: datetime datetime.now()5.2 实现一个简单的执行引擎接下来实现一个能驱动Step执行的引擎。为了简化我们假设 LLM 调用和工具调用都是模拟的。class SimpleAgentEngine: def __init__(self, session: Session): self.session session # 模拟的工具库 self.tools { “get_weather”: self._mock_get_weather, “calculate”: self._mock_calculate, } def run_turn(self, turn: Turn): 执行一个Turn print(f“开始执行 Turn: {turn.id}”) # 1. 创建初始LLM Step current_step self._create_llm_step(turn) turn.steps.append(current_step) while turn.status TurnStatus.RUNNING: # 2. 执行当前Step self._execute_single_step(current_step) # 3. 根据当前Step结果决定下一步 if current_step.status StepStatus.ERROR: turn.status TurnStatus.ERROR break if current_step.type StepType.LLM: # 解析LLM输出判断是否需要工具调用 llm_output current_step.output.get(“content”, “”) if “call_tool” in llm_output: # 简化解析实际应用需要更严谨的解析器 tool_name llm_output.split(“name”)[1].split(“””)[0] # 简陋解析 next_step self._create_tool_call_step(turn, tool_name, llm_output) else: # LLM直接给出了最终答案 turn.response llm_output turn.status TurnStatus.COMPLETED break elif current_step.type StepType.TOOL_CALL: # 工具调用完毕创建结果格式化Step next_step self._create_tool_result_step(turn, current_step.output) elif current_step.type StepType.TOOL_RESULT: # 有了工具结果需要再次询问LLM next_step self._create_llm_step_with_result(turn, current_step.output) else: break # 4. 将新Step加入循环 if next_step: turn.steps.append(next_step) current_step next_step else: break # 5. 将完成的Turn加入Session if turn.status TurnStatus.COMPLETED: self.session.turns.append(turn) print(f“Turn {turn.id} 执行结束状态: {turn.status}”) def _execute_single_step(self, step: Step): 执行单个Step更新其状态和输出 step.status StepStatus.RUNNING step.updated_at datetime.now() try: if step.type StepType.LLM: # 模拟LLM调用根据输入消息生成回复 messages step.input.get(“messages”, []) # 这里应该调用真实的LLM API我们模拟一个 simulated_response self._simulate_llm(messages) step.output {“content”: simulated_response} step.status StepStatus.SUCCESS elif step.type StepType.TOOL_CALL: # 执行工具 tool_name step.input.get(“tool_name”) tool_params step.input.get(“params”, {}) if tool_name in self.tools: result self.tools[tool_name](**tool_params) step.output {“raw_result”: result} step.status StepStatus.SUCCESS else: raise ValueError(f“Tool {tool_name} not found.”) elif step.type StepType.TOOL_RESULT: # 格式化工具结果 raw_result step.input.get(“raw_result”) formatted f“Tool returned: {raw_result}” step.output {“formatted_result”: formatted} step.status StepStatus.SUCCESS except Exception as e: step.status StepStatus.ERROR step.error str(e) step.output None step.updated_at datetime.now() def _simulate_llm(self, messages: List[Dict]) - str: 一个非常简陋的LLM模拟器仅用于演示逻辑 last_msg messages[-1][“content”] if messages else “” if “weather” in last_msg.lower(): return “I need to check the weather. call_tool name\”get_weather\” city\”Beijing\”” elif “calculate” in last_msg.lower(): return “I need to calculate. call_tool name\”calculate\” expression\”22\”” elif “Tool returned” in last_msg: return f“Based on the tool result, the answer is: {last_msg}” else: return “I don’t know how to handle that.” def _mock_get_weather(self, city: str) - str: return f“Sunny, 22°C in {city}” def _mock_calculate(self, expression: str) - str: return str(eval(expression)) # 注意实际生产环境绝对不要用eval # ... 省略 _create_llm_step, _create_tool_call_step 等辅助方法5.3 实现 Session 的保存与加载最后实现一个简单的基于 JSON 文件的SessionStore。import json from pathlib import Path class JsonSessionStore: def __init__(self, storage_dir: Path): self.storage_dir storage_dir storage_dir.mkdir(parentsTrue, exist_okTrue) def save(self, session: Session) - str: 保存Session到JSON文件 file_path self.storage_dir / f“{session.id}.json” # 使用Pydantic的dict()方法进行序列化 session_dict session.dict() # 处理datetime对象使其可JSON序列化Pydantic默认已处理这里确保一下 with open(file_path, ‘w’, encoding‘utf-8’) as f: json.dump(session_dict, f, indent2, defaultstr) return session.id def load(self, session_id: str) - Session: 从JSON文件加载Session file_path self.storage_dir / f“{session_id}.json” if not file_path.exists(): raise FileNotFoundError(f“Session {session_id} not found.”) with open(file_path, ‘r’, encoding‘utf-8’) as f: session_dict json.load(f) # 使用Pydantic的parse_obj方法重建对象 # 注意枚举字段需要特殊处理或者确保Config中use_enum_valuesTrue return Session.parse_obj(session_dict)5.4 运行与验证现在我们可以串联起整个流程# 1. 创建一个新的Session和Turn session Session(id“sess_demo_001”, system_prompt“You are a helpful assistant.”) turn Turn(id“turn_1”, session_idsession.id, prompt“What’s the weather in Beijing?”) # 2. 创建引擎并执行 engine SimpleAgentEngine(session) engine.run_turn(turn) # 3. 打印执行过程 print(f“最终回复: {turn.response}”) print(f“Turn 包含步骤数: {len(turn.steps)}”) for i, step in enumerate(turn.steps): print(f“ Step {i1} [{step.type}]: status{step.status}, output{step.output}”) # 4. 保存Session store JsonSessionStore(Path(“./sessions”)) saved_id store.save(session) print(f“Session 已保存ID: {saved_id}”) # 5. 模拟中断后重新加载并查看 print(“\n--- 模拟进程重启加载Session ---”) loaded_session store.load(saved_id) print(f“加载的Session有 {len(loaded_session.turns)} 个Turns”) if loaded_session.turns: loaded_turn loaded_session.turns[0] print(f“第一个Turn的最终回复: {loaded_turn.response}”) # 理论上我们可以基于loaded_session创建一个新引擎继续执行新的Turn通过这个简易实现你就能清晰地看到Session/Turn/Step如何组织。引擎如何循环执行Step。状态status如何流转。数据如何被序列化和反序列化。这比直接阅读复杂的 Harness 源码更能帮你建立直觉。在实际项目中你可以基于这个骨架替换掉模拟的 LLM 和工具接入真实的模型 API 和业务工具并增强错误处理、上下文构建、更复杂的工具调用解析等功能。6. 总结从 Harness 设计中我们能学到什么DeepSeek Harness 通过Turn/Step/Session这套设计提供了一个关于如何构建可观测、可控制、可持久化 Agent的范本。在你自己设计 Agent 系统时无论是否使用 Harness都可以借鉴这几个核心思想第一将执行过程原子化、状态化。不要只把 Agent 当作一个黑箱函数。把它的一次完整响应拆解成LLM推理-工具调用-结果处理-再推理这样的原子步骤Step并为每个步骤明确记录输入、输出和状态。这是实现可调试性的基础。第二设计清晰的数据层级和生命周期。Session管理宏观任务Turn管理单轮交互Step管理原子操作。每一层都有自己明确的生命周期创建、运行、完成/错误和职责。这让代码结构更清晰也更容易做权限控制、资源隔离和性能监控。第三持久化设计要面向恢复。持久化不是为了存档而是为了能无缝恢复执行。这意味着你保存的不仅仅是输入和最终输出而是整个状态机当前的所有上下文。在序列化时要特别注意外部依赖如网络客户端的重建问题。第四引擎与状态分离。执行引擎Engine是纯逻辑它读取Session/Turn/Step的状态执行操作然后更新状态。状态本身是独立的数据结构。这种分离使得你可以轻松实现不同的持久化后端数据库、文件、内存而不影响核心执行逻辑。最后Harness 的设计也提醒我们复杂的 Agent 工作流本质上是一个状态机。Turn和Step的状态流转就是这个状态机的体现。画清楚你 Agent 可能的状态转换图往往能帮你更好地设计代码结构。如果你正在评估是否要在项目中使用 DeepSeek Harness我的建议是如果你的需求是快速构建一个基于 DeepSeek API 的、需要复杂多步推理和工具调用的 Agent并且非常看重执行过程的透明度和可恢复性那么 Harness 是一个值得深入研究的框架。如果你的 Agent 逻辑非常简单一问一答无需工具或者你已经有一套成熟的异步任务队列和状态管理系统那么你可能只需要借鉴其设计思想而不一定引入整个框架。无论如何理解Turn、Step、Session这套模式都会让你在设计和调试 AI Agent 时更加得心应手。