Agent Plan × DeepSeek Harness 实践指南：从规划到落地的完整代码实战

1. 引言为什么需要 Agent Plan 与 Harness在构建复杂 AI Agent 应用时单纯依赖大模型的即时推理往往难以保证任务执行的稳定性和可追溯性。Agent Plan智能体规划负责将复杂目标拆解为可执行的步骤序列而 Harness执行框架则负责承载这些步骤、管理上下文、调度工具调用并处理异常。两者结合才能构建出真正可靠、可控、可观测的智能体系统。本文将以 DeepSeek 大模型为底座从零开始搭建一套完整的 Agent Plan Harness 实践方案。全文包含可直接运行的 Python 代码、核心概念讲解、架构设计以及常见问题排查帮助你快速掌握这套组合拳的落地方法。2. 核心概念与架构总览在动手写代码之前先厘清几个关键概念这决定了后续代码的组织方式。2.1 Agent Plan 是什么Agent Plan 是智能体对目标任务进行拆解后生成的执行计划通常表现为一个有序的步骤列表。每个步骤包含目标描述、所需工具、输入参数、预期输出以及依赖关系。一个好的 Plan 应当具备以下特征可分解性复杂任务被拆解为粒度适中、可独立执行的子任务。可验证性每个步骤都有明确的完成标准和验证方式。可回溯性步骤之间的依赖关系清晰便于定位失败环节。可扩展性新增任务类型时无需重写整个规划逻辑。2.2 Harness 是什么Harness执行框架是承载 Plan 运行的环境负责以下核心职责上下文管理维护对话历史、中间结果和全局状态。工具调度根据步骤需求调用注册好的外部工具或 API。错误处理捕获异常、重试失败步骤、必要时触发重新规划。观测与日志记录每一步的输入输出便于调试和审计。2.3 整体架构下图展示了本文将要实现的系统架构flowchart TD A[用户输入] -- B[Planner 规划器] B -- C[Plan 步骤列表] C -- D[Harness 执行框架] D -- E[工具调用层] E -- F[DeepSeek API] E -- G[外部工具/API] D -- H[结果汇总] H -- I[最终输出] D -- 步骤失败 -- J[错误处理/重规划] J -- C整个流程可以概括为用户输入目标 → 规划器生成 Plan → Harness 按序执行 → 工具层调用 DeepSeek 或外部服务 → 汇总结果输出。当某一步失败且无法自动恢复时触发重新规划生成新的 Plan 继续执行。3. 环境准备与依赖安装开始编码前先准备好运行环境。本文所有代码基于 Python 3.10建议使用虚拟环境隔离依赖。3.1 安装依赖创建虚拟环境并安装以下核心依赖python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate pip install openai1.35.0 pip install pydantic2.7.0 pip install python-dotenv1.0.1 pip install rich13.7.0说明DeepSeek 的 API 兼容 OpenAI 协议因此我们直接使用 openai 官方 SDK只需修改 base_url 和 api_key 即可。3.2 配置环境变量在项目根目录创建.env文件填入你的 DeepSeek API KeyDEEPSEEK_API_KEYsk-你的密钥 DEEPSEEK_BASE_URLhttps://api.deepseek.com DEEPSEEK_MODELdeepseek-chat注意请勿将.env文件提交到 Git 仓库建议在.gitignore中添加忽略规则。3.3 验证连接先写一个最小脚本验证 DeepSeek API 是否连通import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), ) response client.chat.completions.create( modelos.getenv(DEEPSEEK_MODEL), messages[{role: user, content: 你好请回复连接成功}], temperature0.7, ) print(response.choices[0].message.content)运行后如果输出包含「连接成功」说明环境配置无误可以进入下一步。4. 定义数据模型Plan 与 Step为了让 Plan 结构化、可校验我们使用 Pydantic 定义数据模型。这是整个框架的基石后续的规划器、执行器都围绕这些模型工作。4.1 Step 模型每个步骤包含以下字段from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel, Field class StepStatus(str, Enum): PENDING pending RUNNING running SUCCEEDED succeeded FAILED failed SKIPPED skipped class Step(BaseModel): 单个执行步骤 step_id: str Field(description步骤唯一标识如 step_1) description: str Field(description步骤目标描述) tool: str Field(description执行该步骤所需的工具名称) params: Dict[str, Any] Field(default_factorydict, description工具调用参数) depends_on: List[str] Field(default_factorylist, description依赖的步骤 ID 列表) status: StepStatus StepStatus.PENDING result: Optional[Any] None error: Optional[str] None max_retries: int Field(default2, description最大重试次数) retry_count: int Field(default0, description当前已重试次数)4.2 Plan 模型Plan 是步骤的有序集合同时记录整体状态class Plan(BaseModel): 执行计划 plan_id: str Field(description计划唯一标识) goal: str Field(description总体目标描述) steps: List[Step] Field(description步骤列表) created_at: str Field(description创建时间) status: str Field(defaultcreated, description计划状态: created/running/completed/failed) def get_step(self, step_id: str) - Optional[Step]: 按 ID 获取步骤 for step in self.steps: if step.step_id step_id: return step return None def next_pending_step(self) - Optional[Step]: 获取下一个可执行的待处理步骤依赖已满足 for step in self.steps: if step.status ! StepStatus.PENDING: continue deps_met all( self.get_step(dep).status StepStatus.SUCCEEDED for dep in step.depends_on if self.get_step(dep) is not None ) if deps_met: return step return None这里的关键设计是next_pending_step方法它只返回依赖已全部满足的待执行步骤从而天然支持有向无环图DAG形式的任务依赖关系。5. 实现 Planner让 DeepSeek 生成结构化 PlanPlanner 的核心职责是接收用户目标调用 DeepSeek 生成结构化的步骤列表并解析为 Plan 对象。为了让模型输出稳定可解析的 JSON我们采用「函数调用Function Calling」方式而不是让模型自由输出文本。5.1 定义工具 Schema首先定义 DeepSeek 可以调用的「函数」即生成 Plan 的 JSON Schemaimport json import uuid from datetime import datetime from openai import OpenAI PLAN_TOOL_SCHEMA { type: function, function: { name: generate_plan, description: 根据用户目标生成结构化的执行计划, parameters: { type: object, properties: { steps: { type: array, items: { type: object, properties: { description: {type: string, description: 步骤目标描述}, tool: {type: string, description: 执行工具名称}, params: {type: object, description: 工具参数}, depends_on: { type: array, items: {type: string}, description: 依赖的步骤 ID 列表 } }, required: [description, tool, params, depends_on] } } }, required: [steps] } } }5.2 Planner 类实现接下来实现 Planner 类它负责与 DeepSeek 交互并解析返回的 Planclass Planner: 基于 DeepSeek 的规划器 def __init__(self, client: OpenAI, model: str, available_tools: List[str]): self.client client self.model model self.available_tools available_tools def create_plan(self, goal: str, context: Optional[str] None) - Plan: 根据目标生成执行计划 system_prompt ( 你是一个专业的任务规划器。请将用户的目标拆解为可执行的步骤序列。 f可用的工具包括: {, .join(self.available_tools)}。 注意步骤之间如果有依赖关系请在 depends_on 中明确指定。 步骤数量控制在 3-8 个之间每个步骤要具体、可执行。 ) user_content f目标: {goal} if context: user_content f\n\n补充上下文: {context} messages [ {role: system, content: system_prompt}, {role: user, content: user_content}, ] response self.client.chat.completions.create( modelself.model, messagesmessages, tools[PLAN_TOOL_SCHEMA], tool_choice{type: function, function: {name: generate_plan}}, temperature0.3, ) 解析函数调用参数 tool_calls response.choices[0].message.tool_calls if not tool_calls: raise ValueError(模型未返回有效的 Plan 结构) plan_data json.loads(tool_calls[0].function.arguments) 组装 Step 对象 steps [] for idx, step_data in enumerate(plan_data[steps], start1): step Step( step_idfstep_{idx}, descriptionstep_data[description], toolstep_data[tool], paramsstep_data.get(params, {}), depends_onstep_data.get(depends_on, []), ) steps.append(step) return Plan( plan_idfplan_{uuid.uuid4().hex[:8]}, goalgoal, stepssteps, created_atdatetime.now().isoformat(), )/code/pre 这里的关键点通过 tool_choice 强制模型调用 generate_plan 函数从而保证返回的是严格符合 Schema 的 JSON避免解析失败。 6. 实现 Harness执行框架核心 Harness 是整个系统的执行中枢。它负责按序执行 Plan 中的步骤、调用工具、处理错误、维护上下文并支持失败重试与重新规划。 6.1 工具注册机制 首先实现一个简单的工具注册表让 Harness 能够按名称找到对应的执行函数 from typing import Callable, Dict class ToolRegistry: 工具注册表 def init(self): self._tools: Dict[str, Callable] {} def register(self, name: str, func: Callable): 注册工具 self._tools[name] func def get(self, name: str) - Callable: 获取工具 if name not in self._tools: raise KeyError(f工具 {name} 未注册) return self._tools[name] def list_tools(self) - List[str]: 列出所有已注册工具名 return list(self._tools.keys()) 6.2 内置工具实现 为了让示例可运行我们实现两个内置工具一个调用 DeepSeek 进行文本生成一个模拟外部 API 调用如查询天气 def create_deepseek_tool(client: OpenAI, model: str) - Callable: 创建 DeepSeek 文本生成工具 def deepseek_generate(prompt: str, temperature: float 0.7) - str: response client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperaturetemperature, ) return response.choices[0].message.content return deepseek_generate def create_mock_weather_tool() - Callable: 创建模拟天气查询工具 def get_weather(city: str) - str: 实际项目中这里应调用真实天气 API mock_data { 北京: 晴25°C, 上海: 多云28°C, 广州: 雷阵雨30°C, 深圳: 小雨29°C, } return mock_data.get(city, f{city}暂无数据) return get_weather 6.3 Harness 主类 现在实现 Harness 主类它串联起 Plan 执行、工具调用和错误处理 import time from typing import Any, Dict, Optional class Harness: Agent 执行框架 def init(self, registry: ToolRegistry, planner: Optional[Planner] None): self.registry registry self.planner planner self.context: Dict[str, Any] {} # 全局上下文存储中间结果 self.execution_log: List[Dict] [] # 执行日志 def execute_plan(self, plan: Plan) - Dict[str, Any]: 执行整个 Plan plan.status running self.execution_log.append({event: plan_start, plan_id: plan.plan_id}) while True: step plan.next_pending_step() if step is None: break self._execute_step(plan, step) 如果所有步骤都完成或失败退出循环 all_done all( s.status in (StepStatus.SUCCEEDED, StepStatus.FAILED, StepStatus.SKIPPED) for s in plan.steps ) if all_done: break 汇总状态 failed_steps [s for s in plan.steps if s.status StepStatus.FAILED] if failed_steps: plan.status failed else: plan.status completed self.execution_log.append({event: plan_end, plan_id: plan.plan_id, status: plan.status}) return self._build_result(plan) def _execute_step(self, plan: Plan, step: Step): 执行单个步骤含重试逻辑 step.status StepStatus.RUNNING self.execution_log.append({event: step_start, step_id: step.step_id}) while step.retry_count lt; step.max_retries: try: 解析参数支持从上下文中引用前序步骤结果 resolved_params self._resolve_params(step.params) tool_func self.registry.get(step.tool) result tool_func(**resolved_params) step.status StepStatus.SUCCEEDED step.result result # 将结果存入上下文供后续步骤引用 self.context[step.step_id] result self.execution_log.append({ event: step_success, step_id: step.step_id, result_preview: str(result)[:200], }) return except Exception as e: step.retry_count 1 step.error str(e) self.execution_log.append({ event: step_retry, step_id: step.step_id, attempt: step.retry_count, error: str(e), }) if step.retry_count amp;gt; step.max_retries: step.status StepStatus.FAILED self.execution_log.append({ event: step_failed, step_id: step.step_id, error: str(e), }) # 触发重新规划如果配置了 planner if self.planner: self._replan(plan, step) return time.sleep(1) # 重试前等待 def _resolve_params(self, params: Dict) - Dict: 解析参数将 {{step_id}} 形式的引用替换为上下文中的实际值 resolved {} for key, value in params.items(): if isinstance(value, str) and value.startswith({{) and value.endswith(}}): ref value[2:-2].strip() if ref in self.context: resolved[key] self.context[ref] else: raise ValueError(f上下文不存在引用: {ref}) else: resolved[key] value return resolved def _replan(self, plan: Plan, failed_step: Step): 失败后重新规划基于已完成的步骤和失败原因生成新 Plan self.execution_log.append({ event: replan_triggered, failed_step: failed_step.step_id, reason: failed_step.error, }) 收集已完成步骤的结果摘要 completed_summary \n.join( f- {s.step_id}: {s.description} -gt; {str(s.result)[:100]} for s in plan.steps if s.status StepStatus.SUCCEEDED ) new_plan self.planner.create_plan( goalplan.goal, context( f以下步骤已成功完成:\n{completed_summary}\n f步骤 {failed_step.step_id} 失败原因: {failed_step.error}\n 请基于已完成的工作生成新的计划来完成剩余目标。 ), ) 将新计划并入当前计划简单起见直接替换步骤列表 plan.steps new_plan.steps self.execution_log.append({event: replan_completed, new_step_count: len(new_plan.steps)}) def _build_result(self, plan: Plan) - Dict[str, Any]: 汇总最终结果 return { plan_id: plan.plan_id, goal: plan.goal, status: plan.status, steps: [ { step_id: s.step_id, description: s.description, status: s.status.value, result: s.result, error: s.error, } for s in plan.steps ], context: self.context, } 这段代码是 Harness 的核心有几个设计要点值得注意 参数解析支持 {{step_id}} 语法引用前序步骤的结果实现步骤间数据传递。 重试机制每个步骤独立配置最大重试次数失败后自动重试。 重新规划当某步骤重试耗尽仍失败时调用 Planner 基于已完成进度生成新计划实现动态调整。 执行日志完整记录每个事件便于观测和调试。 7. 完整实战让 Agent 完成一个复合任务 现在把所有模块组装起来运行一个真实的复合任务。我们设计一个场景用户要求「查询北京和上海的天气然后生成一段旅行建议文案」。 7.1 组装系统 import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() 初始化 DeepSeek 客户端 client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), ) model os.getenv(DEEPSEEK_MODEL) 注册工具 registry ToolRegistry() registry.register(deepseek_generate, create_deepseek_tool(client, model)) registry.register(get_weather, create_mock_weather_tool()) 初始化 Planner 和 Harness planner Planner(clientclient, modelmodel, available_toolsregistry.list_tools()) harness Harness(registryregistry, plannerplanner) 生成 Plan goal 查询北京和上海的天气然后基于天气情况生成一段旅行建议文案 plan planner.create_plan(goal) print( 生成的 Plan ) for step in plan.steps: print(f{step.step_id}: {step.description} [工具: {step.tool}] 依赖: {step.depends_on}) 7.2 执行 Plan 执行计划 result harness.execute_plan(plan) print(\n 执行结果 ) print(f计划状态: {result[status]}) for step in result[steps]: status_icon ✅ if step[status] succeeded else ❌ print(f{status_icon} {step[step_id]}: {step[description]}) if step[result]: print(f 结果: {str(step[result])[:150]}) if step[error]: print(f 错误: {step[error]}) 7.3 查看执行日志 print(\n 执行日志 ) for log in harness.execution_log: print(log) 运行这段代码你会看到 Planner 自动生成了类似下面的 Plan step_1: 查询北京的天气 [工具: get_weather] 依赖: [] step_2: 查询上海的天气 [工具: get_weather] 依赖: [] step_3: 基于两地天气生成旅行建议 [工具: deepseek_generate] 依赖: [step_1, step_2] 注意第三步的 depends_on 自动指向了前两个步骤这正是 Planner 理解依赖关系的体现。执行时Harness 会先并行执行 step_1 和 step_2因为互不依赖然后执行 step_3并通过 {{step_1}} 和 {{step_2}} 引用前序结果。 8. 进阶支持并行执行与流式输出 上面的实现是串行执行对于无依赖的步骤可以并行以提升效率。下面给出一个简单的并行执行扩展。 8.1 并行执行无依赖步骤 from concurrent.futures import ThreadPoolExecutor, as_completed class ParallelHarness(Harness): 支持并行执行的 Harness def execute_plan(self, plan: Plan) - Dict[str, Any]: plan.status running self.execution_log.append({event: plan_start, plan_id: plan.plan_id}) with ThreadPoolExecutor(max_workers4) as executor: while True: 找出所有可并行执行的待处理步骤 ready_steps [] for step in plan.steps: if step.status ! StepStatus.PENDING: continue deps_met all( plan.get_step(dep).status StepStatus.SUCCEEDED for dep in step.depends_on if plan.get_step(dep) is not None ) if deps_met: ready_steps.append(step) if not ready_steps: break # 并行执行 futures { executor.submit(self._execute_step, plan, step): step for step in ready_steps } for future in as_completed(futures): future.result() # 触发异常传播 汇总状态与父类相同 failed_steps [s for s in plan.steps if s.status StepStatus.FAILED] plan.status failed if failed_steps else completed self.execution_log.append({event: plan_end, plan_id: plan.plan_id, status: plan.status}) return self._build_result(plan)/code/pre 这个并行版本通过 ThreadPoolExecutor 同时执行所有无依赖的步骤显著缩短整体耗时。注意线程安全由于每个步骤独立写入自己的 step.result 和 self.context[step.step_id]不存在共享可变状态的竞争问题。 8.2 流式输出 DeepSeek 结果 对于耗时较长的文本生成步骤流式输出能显著改善用户体验。下面改造 DeepSeek 工具支持流式 def create_deepseek_stream_tool(client: OpenAI, model: str) - Callable: 创建支持流式输出的 DeepSeek 工具 def deepseek_stream(prompt: str, temperature: float 0.7) - str: stream client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], temperaturetemperature, streamTrue, ) collected [] for chunk in stream: if chunk.choices[0].delta.content: piece chunk.choices[0].delta.content collected.append(piece) print(piece, end, flushTrue) # 实时打印 print() # 换行 return .join(collected) return deepseek_stream 将注册工具改为 registry.register(deepseek_generate, create_deepseek_stream_tool(client, model)) 即可启用流式输出。 9. 错误处理与重新规划实战 真实场景中工具调用难免失败。下面演示如何触发重新规划机制。我们故意让某个工具抛异常观察 Harness 的行为。 9.1 模拟失败场景 def create_flaky_tool(success_rate: float 0.3) - Callable: 创建一个有概率失败的工具用于演示重试和重新规划 import random def flaky_operation(task: str) - str: if random.random() success_rate: raise RuntimeError(f模拟失败: 无法完成 {task}) return f成功完成: {task} return flaky_operation 注册一个不稳定的工具 registry.register(flaky_tool, create_flaky_tool(success_rate0.3)) 构造一个依赖该工具的 Plan plan Plan( plan_idplan_demo_failure, goal演示失败重试与重新规划, steps[ Step(step_idstep_1, description执行不稳定操作, toolflaky_tool, params{task: 数据清洗}, max_retries2), Step(step_idstep_2, description基于结果生成报告, tooldeepseek_generate, params{prompt: {{step_1}} 请基于此生成简短报告}, depends_on[step_1]), ], created_atdatetime.now().isoformat(), ) result harness.execute_plan(plan) print(f最终状态: {result[status]}) for step in result[steps]: print(f{step[step_id]}: {step[status]} | 错误: {step[error]}) 运行多次你会看到两种典型结果 如果 step_1 在重试内成功则 step_2 正常执行计划完成。 如果 step_1 重试耗尽仍失败Harness 会调用 Planner 重新规划生成新的步骤序列来绕过失败点。 重新规划是 Agent 系统智能性的重要体现它不会因为单点失败而整体崩溃而是动态调整策略继续推进目标。 10. 观测与日志最佳实践 生产环境中可观测性是 Agent 系统的生命线。下面给出几个增强观测能力的建议。 10.1 结构化日志 将执行日志输出为 JSON 格式便于接入日志平台 import json import logging logger logging.getLogger(agent_harness) logger.setLevel(logging.INFO) handler logging.StreamHandler() handler.setFormatter(logging.Formatter(%(asctime)s %(message)s)) logger.addHandler(handler) def log_event(event: Dict): 输出结构化日志 logger.info(json.dumps(event, ensure_asciiFalse, defaultstr)) 在 Harness 的各个事件点调用 log_event 替代直接 append 到列表即可同时获得内存日志和标准输出日志。 10.2 执行耗时统计 为每个步骤记录耗时便于定位性能瓶颈 import time 在 _execute_step 中 start_time time.time() ... 执行逻辑 ... elapsed time.time() - start_time self.execution_log.append({ event: step_completed, step_id: step.step_id, elapsed_seconds: round(elapsed, 3), }) 10.3 上下文快照 在关键节点保存上下文快照便于事后复盘 def snapshot_context(self, tag: str): 保存上下文快照 self.execution_log.append({ event: context_snapshot, tag: tag, context: {k: str(v)[:200] for k, v in self.context.items()}, }) 11. 常见问题与排查指南 实践过程中你可能会遇到以下典型问题这里给出排查思路。 11.1 DeepSeek 返回的 Plan 解析失败 现象json.loads 抛出 JSONDecodeError。 原因模型偶尔会返回不规范的 JSON或包含多余的前后文本。 解决 在解析前先提取 JSON 片段如用正则匹配 {.*}。 降低 temperature 到 0.2 以下提高输出稳定性。 增加重试解析失败时重新调用一次 Planner。 11.2 步骤间参数引用失效 现象_resolve_params 抛出「上下文不存在引用」。 原因步骤的 depends_on 未正确声明导致执行顺序错误。 解决检查 Plan 中步骤的 depends_on 字段确保引用前序步骤 ID 正确。 11.3 工具调用超时 现象DeepSeek API 请求长时间无响应。 解决在 OpenAI 客户端初始化时设置超时参数 client OpenAI( api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), timeout30.0, # 30 秒超时 max_retries2, # 自动重试 ) 11.4 重新规划陷入死循环 现象Plan 反复失败、反复重新规划无法收敛。 解决为重新规划设置最大次数限制超过后直接标记失败 在 Harness 中增加 self.max_replans 3 self.replan_count 0 在 _replan 中 if self.replan_count self.max_replans: self.execution_log.append({event: replan_limit_reached}) return self.replan_count 1 12. 总结与扩展方向 本文从零构建了一套完整的 Agent Plan DeepSeek Harness 实践方案涵盖 数据模型用 Pydantic 定义 Step 和 Plan保证结构化和可校验。 Planner通过 Function Calling 让 DeepSeek 生成稳定可解析的 Plan。 Harness实现步骤调度、参数解析、重试、重新规划和日志观测。 并行与流式扩