DeepSeek Harness：智能体状态管理的核心原理与工程实践

在实际智能体开发中，一个长期困扰开发者的核心问题是：智能体的状态（如对话历史、用户偏好、任务上下文）应该存储在哪里？是放在前端浏览器的内存里，还是后端服务器的某个全局变量里？当智能体需要跨会话、跨设备、甚至跨平台保持连续性时，这种“状态归属”的模糊性会直接导致数据丢失、会话混乱和难以调试。DeepSeek Harness 正是为了解决这一问题而设计的框架，它明确提出了“状态有明确归属”的设计哲学，旨在为智能体提供一个清晰、可管理、可持久化的状态管理方案。

本文面向正在或计划使用 DeepSeek 等大模型 API 构建复杂智能体应用的开发者。我们将从零开始，理解 DeepSeek Harness 的核心概念，搭建一个具备状态管理能力的智能体，并深入探讨其背后的设计原理、实现细节以及生产环境下的最佳实践。通过本文，你将掌握如何构建一个状态清晰、可回溯、且易于扩展的智能体系统。

1. 理解 DeepSeek Harness 的核心：状态归属与智能体生命周期

在深入代码之前，必须厘清两个核心概念：状态归属与智能体生命周期。这是理解 DeepSeek Harness 设计意图的基石。

1.1 为什么智能体状态需要明确归属？

传统的、简单的聊天机器人实现，状态管理往往是混乱的。常见的问题模式包括：

• 内存状态：将对话历史存储在服务器的内存（如一个全局字典）中。服务器重启，对话清零；用户量增大，内存暴涨。
• 无状态设计：每次请求都携带完整的上下文。这虽然符合 RESTful 无状态原则，但对于长对话，每次传输大量历史记录，效率低下，且客户端负担重。
• 混合状态：部分状态在前端，部分在后端，缺乏统一的同步和持久化机制，导致调试时如同“黑盒”。

DeepSeek Harness 倡导的“状态有明确归属”，是指智能体的每一次交互、每一个决策所依赖的上下文数据，都应该被清晰地定义、存储和管理。这个“归属地”就是Harness。你可以将 Harness 理解为一个智能体的运行时容器或会话管理器。它负责：

1. 创建和维持一个智能体实例的完整生命周期。
2. 托管该智能体的所有状态数据（如对话记忆、工具调用历史、用户配置）。
3. 提供接口供外部系统（如 Web 服务器、消息队列消费者）与智能体进行交互。
4. 管理状态持久化，确保智能体状态不因进程重启而丢失。

1.2 智能体生命周期的 Harness 视角

一个由 Harness 管理的智能体，其生命周期通常遵循以下流程：

1. 创建 (Create)：根据配置（模型、系统提示词、初始参数）创建一个智能体实例，并为其分配一个唯一的会话 ID (session_id)。
2. 运行/交互 (Run/Interact)：外部请求通过session_id找到对应的 Harness 和智能体，传入用户输入。Harness 负责加载该智能体的状态，执行推理（可能调用工具），更新状态（如追加对话历史），并返回响应。
3. 持久化 (Persist)：在关键节点（如每次交互后、或定时）将智能体的状态序列化并存储到数据库或文件中。
4. 销毁/归档 (Destroy/Archive)：当会话结束（如用户长时间不活跃），可以安全地销毁 Harness 实例，但其状态数据仍被持久化，未来可通过session_id恢复。

这种设计使得智能体不再是“一次性的函数调用”，而是一个有状态的、可长期运行的、可管理的实体。

2. 环境准备与项目初始化

我们将使用 Python 作为开发语言，因为它拥有最丰富的 AI 开发生态。DeepSeek Harness 的核心思想是框架无关的，你可以用任何语言实现其理念。这里我们用一个模拟的 Harness 框架结构来演示。

2.1 环境与依赖

首先，确保你的 Python 环境版本在 3.8 及以上。我们主要需要以下库：

# 创建项目目录并进入 mkdir deepseek_agent_harness && cd deepseek_agent_harness # 创建虚拟环境（推荐） python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 安装核心依赖 pip install openai # 用于调用 DeepSeek API (兼容 OpenAI SDK) pip install pydantic # 用于数据验证和设置管理 pip install redis # 可选，用于演示分布式状态存储 pip install sqlalchemy # 可选，用于演示数据库状态存储 pip install fastapi uvicorn # 可选，用于构建 Web API 服务

注意：截至撰写时，DeepSeek 的 API 与 OpenAI SDK 兼容。因此，我们可以直接使用openai这个官方库，只需将base_url和api_key替换为 DeepSeek 的端点。请确保你已从 DeepSeek 平台获取有效的 API Key。

2.2 项目结构设计

一个清晰的项目结构是管理复杂状态的基础。建议如下：

deepseek_agent_harness/ ├── app/ │ ├── __init__.py │ ├── core/ │ │ ├── __init__.py │ │ ├── config.py # 配置文件 │ │ ├── state.py # 状态数据模型定义 │ │ └── harness.py # Harness 核心类 │ ├── agents/ │ │ ├── __init__.py │ │ └── deepseek_agent.py # 具体的智能体实现 │ ├── storage/ │ │ ├── __init__.py │ │ ├── base.py # 存储抽象接口 │ │ ├── memory.py # 内存存储（用于开发） │ │ └── redis_store.py # Redis 存储实现 │ └── api/ │ ├── __init__.py │ └── server.py # FastAPI 服务入口 ├── requirements.txt └── .env.example # 环境变量示例

这个结构将核心的 Harness 逻辑、智能体定义、状态存储和对外 API 分离开，符合“状态有明确归属”的理念，每个模块职责清晰。

3. 实现核心：定义状态与构建 Harness

现在，我们从内向外构建。首先定义智能体的状态，然后实现管理这个状态的 Harness。

3.1 定义智能体状态模型

在app/core/state.py中，我们使用 Pydantic 来定义状态。Pydantic 提供了数据验证和序列化能力，非常适合用来描述状态结构。

from datetime import datetime from typing import List, Dict, Any, Optional from pydantic import BaseModel, Field class Message(BaseModel): """表示对话中的一条消息""" role: str # 'system', 'user', 'assistant', 'tool' content: str name: Optional[str] = None # 可选，工具调用时的函数名 tool_calls: Optional[List[Dict]] = None # 模型请求调用工具的信息 tool_call_id: Optional[str] = None # 工具调用的ID，用于匹配结果 class AgentState(BaseModel): """智能体的完整状态""" session_id: str = Field(..., description="会话的唯一标识符") created_at: datetime = Field(default_factory=datetime.now) updated_at: datetime = Field(default_factory=datetime.now) # 核心：对话历史 message_history: List[Message] = Field(default_factory=list) # 其他自定义状态 user_metadata: Dict[str, Any] = Field(default_factory=dict) # 用户偏好、身份等 agent_config: Dict[str, Any] = Field(default_factory=dict) # 本次会话的特定配置 # 例如：当前任务阶段、已收集的信息、临时变量等 context: Dict[str, Any] = Field(default_factory=dict) class Config: # 允许任意类型，方便扩展 arbitrary_types_allowed = True def update_timestamp(self): """更新状态时间戳""" self.updated_at = datetime.now() def add_message(self, message: Message): """向历史中添加消息，并更新状态""" self.message_history.append(message) self.update_timestamp() def to_dict(self) -> Dict[str, Any]: """序列化为字典，便于存储""" return self.dict() @classmethod def from_dict(cls, data: Dict[str, Any]) -> "AgentState": """从字典反序列化""" return cls(**data)

这个AgentState类就是智能体状态的“明确归属”。所有与本次会话相关的数据都封装在此。

3.2 实现 Harness 核心类

接下来，在app/core/harness.py中实现 Harness。Harness 的核心职责是绑定一个智能体逻辑与其状态，并提供执行入口。

import asyncio import logging from typing import Any, Callable, Optional from .state import AgentState logger = logging.getLogger(__name__) class Harness: """ Harness 核心类。 1. 持有智能体的状态 (AgentState)。 2. 执行智能体的处理逻辑。 3. 委托存储层进行状态的持久化与加载。 """ def __init__( self, session_id: str, agent_logic: Callable, # 智能体的核心处理函数 storage_backend: Any, # 存储后端实例 initial_state: Optional[Dict[str, Any]] = None ): self.session_id = session_id self.agent_logic = agent_logic self.storage = storage_backend # 加载或初始化状态 self._state: Optional[AgentState] = None self._load_or_init_state(initial_state or {}) def _load_or_init_state(self, initial_data: Dict[str, Any]): """从存储加载状态，若不存在则初始化""" try: stored_data = self.storage.load(self.session_id) if stored_data: self._state = AgentState.from_dict(stored_data) logger.info(f"Session {self.session_id} state loaded from storage.") else: # 初始化新状态 self._state = AgentState(session_id=self.session_id, **initial_data) logger.info(f"Session {self.session_id} state initialized.") except Exception as e: logger.error(f"Failed to load state for session {self.session_id}: {e}") # 降级：使用初始数据创建新状态 self._state = AgentState(session_id=self.session_id, **initial_data) async def run(self, user_input: str, **kwargs) -> str: """ 执行一次智能体交互。 1. 将用户输入添加到状态。 2. 调用智能体逻辑处理。 3. 更新状态（包括AI回复）。 4. 持久化状态。 5. 返回AI回复。 """ if self._state is None: raise RuntimeError("Agent state not initialized.") from .state import Message # 1. 更新状态：添加用户消息 user_message = Message(role="user", content=user_input) self._state.add_message(user_message) # 2. 调用智能体逻辑（这是一个异步函数） # 我们将当前状态和会话ID传递给智能体 try: agent_response = await self.agent_logic( state=self._state, session_id=self.session_id, **kwargs ) except Exception as e: logger.error(f"Agent logic failed for session {self.session_id}: {e}") # 可以在这里添加一个错误消息到状态 error_message = Message(role="assistant", content=f"处理请求时发生错误：{str(e)}") self._state.add_message(error_message) agent_response = error_message.content # 3. 更新状态：添加AI回复消息 (假设agent_response是Message或字符串) if isinstance(agent_response, Message): assistant_message = agent_response else: assistant_message = Message(role="assistant", content=str(agent_response)) self._state.add_message(assistant_message) # 4. 持久化状态 (异步保存) try: # 注意：生产环境可能需要考虑更细粒度的锁或乐观锁 await self.storage.save(self.session_id, self._state.to_dict()) except Exception as e: logger.error(f"Failed to persist state for session {self.session_id}: {e}") # 持久化失败不应影响本次响应，但需要告警 # 5. 返回响应内容 return assistant_message.content def get_state(self) -> Optional[AgentState]: """获取当前状态（只读）""" return self._state async def close(self): """关闭 Harness，释放资源""" # 确保最终状态被保存 if self._state: try: await self.storage.save(self.session_id, self._state.to_dict()) except Exception as e: logger.error(f"Final save failed on close for session {self.session_id}: {e}") logger.info(f"Harness for session {self.session_id} closed.")

这个Harness类是一个通用的状态管理器。它不关心智能体具体用什么模型（DeepSeek 或其他），只关心如何管理AgentState和执行agent_logic。

4. 集成 DeepSeek 并实现智能体逻辑

现在，我们创建一个具体的智能体，它使用 DeepSeek API 进行对话。

4.1 配置与客户端初始化

在app/core/config.py中管理配置：

import os from pydantic_settings import BaseSettings class Settings(BaseSettings): # DeepSeek API 配置 DEEPSEEK_API_KEY: str = os.getenv("DEEPSEEK_API_KEY", "") DEEPSEEK_BASE_URL: str = "https://api.deepseek.com" # 以官方最新文档为准 DEEPSEEK_MODEL: str = "deepseek-chat" # 例如 deepseek-chat, deepseek-coder # 应用配置 STATE_STORAGE_TYPE: str = os.getenv("STATE_STORAGE_TYPE", "memory") # memory, redis # Redis 配置 (如果使用) REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0") class Config: env_file = ".env" settings = Settings()

在app/agents/deepseek_agent.py中实现智能体逻辑：

import openai import logging from typing import Dict, Any from app.core.config import settings from app.core.state import AgentState, Message logger = logging.getLogger(__name__) # 初始化 OpenAI 客户端（兼容 DeepSeek） client = openai.OpenAI( api_key=settings.DEEPSEEK_API_KEY, base_url=settings.DEEPSEEK_BASE_URL, ) async def deepseek_agent_logic(state: AgentState, session_id: str, **kwargs) -> str: """ 智能体核心逻辑。 1. 从 state 中提取对话历史。 2. 调用 DeepSeek API。 3. 处理可能的工具调用（此处简化，未实现）。 4. 返回助理消息。 """ if not settings.DEEPSEEK_API_KEY: return "错误：未配置 DeepSeek API Key。" # 1. 准备 API 调用所需的 messages 格式 # 通常我们会保留全部历史，但长上下文模型有token限制，需要做摘要或截断。 # 这里简单地将所有消息转换为 API 格式。 messages_for_api = [] for msg in state.message_history[-20:]: # 简单限制历史长度，防止超出token限制 api_msg = {"role": msg.role, "content": msg.content} if msg.name: api_msg["name"] = msg.name if msg.tool_calls: api_msg["tool_calls"] = msg.tool_calls if msg.tool_call_id: api_msg["tool_call_id"] = msg.tool_call_id messages_for_api.append(api_msg) # 2. 调用 DeepSeek API try: response = client.chat.completions.create( model=settings.DEEPSEEK_MODEL, messages=messages_for_api, stream=False, # 简化处理，先不使用流式 # 可以在此添加 temperature, max_tokens 等参数 **kwargs.get('model_params', {}) ) assistant_message_content = response.choices[0].message.content # 注意：实际响应中可能包含 tool_calls，这里简化处理，只返回文本内容 return assistant_message_content except openai.APIError as e: logger.error(f"DeepSeek API call failed for session {session_id}: {e}") return f"调用AI服务时遇到问题：{e.message}" except Exception as e: logger.error(f"Unexpected error in agent logic for session {session_id}: {e}") return "智能体处理过程中发生未知错误。"

4.2 实现状态存储层

Harness 需要存储后端。我们先实现一个内存存储用于开发，再实现一个 Redis 存储用于演示生产环境。

在app/storage/base.py中定义接口：

from abc import ABC, abstractmethod from typing import Optional, Dict, Any class StateStorage(ABC): """状态存储抽象基类""" @abstractmethod async def save(self, session_id: str, state_data: Dict[str, Any]) -> bool: """保存状态数据""" pass @abstractmethod async def load(self, session_id: str) -> Optional[Dict[str, Any]]: """加载状态数据""" pass @abstractmethod async def delete(self, session_id: str) -> bool: """删除状态数据""" pass

在app/storage/memory.py中实现内存存储：

import asyncio from typing import Optional, Dict, Any from .base import StateStorage class MemoryStorage(StateStorage): """内存存储，仅用于开发和测试""" def __init__(self): self._storage: Dict[str, Dict[str, Any]] = {} self._lock = asyncio.Lock() async def save(self, session_id: str, state_data: Dict[str, Any]) -> bool: async with self._lock: self._storage[session_id] = state_data return True async def load(self, session_id: str) -> Optional[Dict[str, Any]]: async with self._lock: return self._storage.get(session_id) async def delete(self, session_id: str) -> bool: async with self._lock: if session_id in self._storage: del self._storage[session_id] return True return False

在app/storage/redis_store.py中实现 Redis 存储：

import json import asyncio from typing import Optional, Dict, Any import redis.asyncio as redis from .base import StateStorage from app.core.config import settings class RedisStorage(StateStorage): """Redis 存储，适用于生产环境""" def __init__(self, redis_url: str = settings.REDIS_URL, ttl: int = 86400): # TTL: 状态过期时间（秒），例如 24小时 self.redis_url = redis_url self.ttl = ttl self._client: Optional[redis.Redis] = None async def _get_client(self) -> redis.Redis: """获取 Redis 客户端（懒加载）""" if self._client is None: self._client = redis.from_url(self.redis_url, decode_responses=True) return self._client async def save(self, session_id: str, state_data: Dict[str, Any]) -> bool: try: client = await self._get_client() # 使用 JSON 序列化状态数据 serialized = json.dumps(state_data, default=str) # default=str 处理 datetime await client.setex(f"agent_state:{session_id}", self.ttl, serialized) return True except Exception as e: print(f"Redis save error: {e}") return False async def load(self, session_id: str) -> Optional[Dict[str, Any]]: try: client = await self._get_client() data = await client.get(f"agent_state:{session_id}") if data: return json.loads(data) return None except Exception as e: print(f"Redis load error: {e}") return None async def delete(self, session_id: str) -> bool: try: client = await self._get_client() result = await client.delete(f"agent_state:{session_id}") return result > 0 except Exception as e: print(f"Redis delete error: {e}") return False async def close(self): if self._client: await self._client.close()

5. 组装与运行：创建完整的智能体服务

现在，我们将所有部分组合起来，并通过一个 Web API 提供服务。

5.1 创建 Harness 管理器

我们需要一个管理器来创建和获取 Harness 实例，避免为同一会话重复创建。

在app/core/harness_manager.py中：

import asyncio import logging from typing import Dict, Optional from .harness import Harness from app.storage.memory import MemoryStorage from app.storage.redis_store import RedisStorage from app.core.config import settings from app.agents.deepseek_agent import deepseek_agent_logic logger = logging.getLogger(__name__) class HarnessManager: """管理 Harness 实例的生命周期""" _instances: Dict[str, Harness] = {} _lock = asyncio.Lock() @classmethod def _get_storage_backend(cls): """根据配置获取存储后端""" if settings.STATE_STORAGE_TYPE.lower() == "redis": return RedisStorage() else: # 默认使用内存存储 return MemoryStorage() @classmethod async def get_harness(cls, session_id: str, create_if_missing: bool = True) -> Optional[Harness]: """ 获取指定 session_id 的 Harness。 如果不存在且 create_if_missing 为 True，则创建一个新的。 """ async with cls._lock: harness = cls._instances.get(session_id) if harness is not None: return harness if not create_if_missing: return None # 创建新的 Harness storage = cls._get_storage_backend() # 可以在这里传递初始状态，例如从数据库加载用户信息 initial_state = { "user_metadata": {"source": "api"}, "agent_config": {"max_history_length": 20} } harness = Harness( session_id=session_id, agent_logic=deepseek_agent_logic, storage_backend=storage, initial_state=initial_state ) cls._instances[session_id] = harness logger.info(f"Created new Harness for session: {session_id}") return harness @classmethod async def cleanup_session(cls, session_id: str): """清理并关闭一个会话的 Harness""" async with cls._lock: harness = cls._instances.pop(session_id, None) if harness: await harness.close() logger.info(f"Cleaned up Harness for session: {session_id}") @classmethod async def cleanup_all(cls): """清理所有 Harness 实例（例如在服务关闭时）""" async with cls._lock: for session_id, harness in list(cls._instances.items()): await harness.close() cls._instances.clear() logger.info("Cleaned up all Harness instances.")

5.2 构建 FastAPI Web 服务

在app/api/server.py中：

from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel import logging from app.core.harness_manager import HarnessManager logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="DeepSeek Agent Harness API") class ChatRequest(BaseModel): message: str session_id: str # 客户端需要管理并传递 session_id # 可选：可以传递模型参数覆盖默认值 model_params: dict = {} class ChatResponse(BaseModel): reply: str session_id: str @app.post("/chat", response_model=ChatResponse) async def chat_with_agent(request: ChatRequest): """ 与智能体对话的端点。 客户端必须提供 session_id 以维持会话状态。 """ if not request.session_id: raise HTTPException(status_code=400, detail="session_id is required") if not request.message.strip(): raise HTTPException(status_code=400, detail="message cannot be empty") try: # 1. 获取或创建该会话的 Harness harness = await HarnessManager.get_harness(request.session_id) if not harness: raise HTTPException(status_code=500, detail="Failed to initialize agent harness") # 2. 运行智能体 reply = await harness.run(request.message, **request.model_params) # 3. 返回响应 return ChatResponse(reply=reply, session_id=request.session_id) except Exception as e: logger.error(f"Error processing chat for session {request.session_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.on_event("shutdown") async def shutdown_event(): """服务关闭时，清理所有 Harness 资源""" await HarnessManager.cleanup_all() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

5.3 运行与测试

1. 准备环境变量：创建.env文件（参考.env.example）。

   DEEPSEEK_API_KEY=your_deepseek_api_key_here STATE_STORAGE_TYPE=memory # 先用内存存储测试

2. 启动服务：

   cd deepseek_agent_harness python -m app.api.server

3. 测试 API：使用curl或 Postman 进行测试。

   # 第一次请求，创建新会话 curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{ "session_id": "test_user_001", "message": "你好，请介绍一下你自己。" }' # 使用相同的 session_id 继续对话，智能体会记住上下文 curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{ "session_id": "test_user_001", "message": "我上一个问题是什么？" }' # 新会话，状态独立 curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{ "session_id": "another_user_002", "message": "我们刚才聊过天吗？" }'

6. 生产环境考量与最佳实践

上述示例是一个可运行的最小化版本。要将它用于生产，必须考虑以下方面：

6.1 状态存储选型与优化

最佳实践：

• 读写分离：高频的load/save操作使用 Redis，定期将完整状态快照持久化到数据库。
• 状态分片：对于超长对话，不要将整个历史都存入一个状态。可以将历史消息单独存储（如时序数据库），状态中只保留摘要或指针。
• 压缩与序列化：状态 JSON 可能很大，考虑使用msgpack或orjson替代json，并在存储前用zlib压缩。

6.2 并发、锁与一致性

当多个请求同时操作同一个session_id时，会出现状态竞争。

• 问题：请求 A 加载状态，请求 B 也加载了相同的状态。A 处理完保存，B 随后也保存，会覆盖 A 的更改。
• 解决方案：
  1. 会话锁：在HarnessManager.get_harness和harness.run层面，对同一个session_id的请求进行排队（如使用asyncio.Lock字典）。这会影响吞吐量。
  2. 乐观锁：在AgentState中增加一个version字段。加载时记录版本号，保存时检查版本号是否变化。如果变化，则重试或报错。这需要存储层支持原子比较和设置（如 Redis 的WATCH/MULTI/EXEC或SETNX）。
  3. 最终一致性：对于某些对话场景，允许短暂的状态不一致，通过后续的对话自动修正。这需要业务能容忍。

推荐实现（乐观锁示例）： 在AgentState中增加version: int = 0。 在RedisStorage.save中：

async def save(self, session_id: str, state_data: Dict[str, Any]) -> bool: async with self._client.pipeline(transaction=True) as pipe: try: await pipe.watch(f"agent_state:{session_id}") current_data = await pipe.get(f"agent_state:{session_id}") current_version = json.loads(current_data).get('version', 0) if current_data else 0 incoming_version = state_data.get('version', 0) if current_version != incoming_version: await pipe.unwatch() return False # 版本冲突，保存失败 # 版本号递增 state_data['version'] = incoming_version + 1 serialized = json.dumps(state_data, default=str) pipe.multi() pipe.setex(f"agent_state:{session_id}", self.ttl, serialized) await pipe.execute() return True except redis.WatchError: return False # 在监视期间键被修改，保存失败

6.3 监控、日志与排查

清晰的日志是排查状态相关问题的关键。

• 结构化日志：使用structlog或json-logging，在每条日志中记录session_id。
• 关键事件打点：在状态加载、保存、版本冲突、存储失败时记录日志。
• 状态快照：在发生难以复现的错误时，可以将有问题的状态快照保存到独立文件或诊断存储中，便于离线分析。
• 指标监控：监控 Harness 创建数、状态保存成功率、平均状态大小、Redis 内存使用量等。

6.4 常见问题排查表

7. 扩展方向

基于这个清晰的 Harness 框架，你可以轻松扩展智能体的能力：

1. 工具调用集成：在AgentState中增加tool_execution_history字段。在agent_logic中解析模型的tool_calls响应，调用相应函数，并将结果以tool角色的消息格式追加到message_history。
2. 长期记忆与摘要：当message_history过长时，可以触发一个摘要过程，将早期对话总结成一段文本，替换掉原始消息，从而节省 token 并保留关键信息。
3. 多模态支持：扩展Message模型和AgentState，支持图像、文档等输入。在调用 API 前，将多媒体内容处理成符合 DeepSeek API 要求的格式（如 base64）。
4. 工作流与状态机：在AgentState的context字段中定义当前任务阶段。agent_logic根据阶段选择不同的系统提示词或工具集，实现复杂的多轮任务自动化。
5. 分布式部署：将HarnessManager和状态存储（Redis）独立出来，Web API 服务可以水平扩展。确保同一session_id的请求通过负载均衡器（如 Nginx 的ip_hash）或分布式会话方案路由到同一后端实例。

通过 DeepSeek Harness 这种将状态管理抽象出来的设计，智能体的核心逻辑（与大模型交互）与状态持久化、会话管理解耦。这使得你的智能体应用更容易测试、调试和扩展。状态有了明确的归属，不再是散落在各处的隐式变量，而是成为了你系统中一个一等公民，可以被观察、管理和优化。