在实际 AI 应用开发中我们经常听到 Skill、Agent、Harness、Subagent 这些术语它们频繁出现在各类框架文档和项目讨论里。很多开发者初次接触时会感到困惑它们各自代表什么在技术架构中扮演什么角色如何配合起来完成一个复杂的任务如果只是孤立地看每个概念的定义很容易陷入“懂每个词但不知道如何组装”的困境。本文将从工程实践的角度通过构建一个模拟的“技术文章助手”项目拆解这四个核心组件的职责、交互关系与具体应用方式。我们将遵循“概念解释 - 项目设计 - 环境搭建 - 核心实现 - 运行验证 - 问题排查”的路径让你不仅能理解理论更能掌握如何在自己的项目中应用这套架构模式。1. 理解核心概念从抽象定义到具体职责在开始编码之前必须厘清每个术语在技术上下文中的具体含义。它们并非互斥而是描述了同一系统中不同层次和粒度的组件。1.1 Agent系统的“大脑”与决策中心Agent 是一个具有自主性、能感知环境、做出决策并执行行动以达成目标的实体。在软件工程中尤其是在 AI 驱动的应用中Agent 通常指一个封装了目标、记忆、规划能力和工具调用能力的程序模块。它不一定是“智能”的但一定是“主动”的。一个 Agent 的核心职责包括目标理解解析用户或上级指令明确要完成的任务。任务规划将复杂目标拆解为一系列可执行的子任务或步骤。工具协调决定在哪个步骤调用哪个工具Skill或委托给哪个下属Subagent。状态管理维护任务执行的上下文、历史记录和中间结果。决策与迭代根据执行结果判断是否继续、重试或调整计划。在我们的“技术文章助手”项目中主 Agent 就是一个TechnicalWritingAgent它的目标是“根据用户主题生成一篇结构完整的技术博客草稿”。它自己并不直接写代码或查资料而是负责规划“先查资料、再列大纲、然后写正文、最后检查格式”这个流程。1.2 SkillAgent 的“手”与基础能力单元Skill 是 Agent 能够调用的具体、原子化的能力或工具。一个 Skill 通常完成一件非常明确的事情例如“调用搜索引擎 API”、“执行一个 Python 函数”、“读写特定数据库”或“调用大语言模型生成文本”。Skill 的设计原则是高内聚、低耦合它不应该包含复杂的业务逻辑或决策过程。Agent 通过组合和编排不同的 Skills 来完成复杂任务。关键区别在于Agent 决定“做什么”和“何时做”Skill 负责“怎么做”。例如我们的TechnicalWritingAgent可能需要以下 SkillsWebSearchSkill: 根据关键词从互联网获取最新技术资料。CodeAnalysisSkill: 解析给定的代码片段提取函数签名和注释。TextGenerationSkill: 调用 LLM API根据提示词生成连贯的文本段落。MarkdownFormatSkill: 将纯文本内容格式化为符合 Markdown 规范的文档。1.3 HarnessAgent 的“运行环境”与生命周期管理器Harness 的本意是“马具”引申为“控制或利用某物的装置”。在 Agent 架构中Harness 指的是承载、启动、监控和管理 Agent 生命周期的基础设施或框架。它提供了 Agent 运行所需的通用服务例如消息路由在多个 Agent 或 Skill 之间传递请求和响应。状态持久化保存和加载 Agent 的对话历史、执行状态。工具注册与发现管理所有可用的 Skills供 Agent 查询和调用。并发与超时控制管理 Agent 任务的执行线程和超时策略。日志与可观测性统一收集运行日志、性能指标和追踪信息。你可以将 Harness 理解为类似 Spring 之于 Java 应用的角色。它不关心具体的业务逻辑那是 Agent 的事但提供了业务逻辑运行所依赖的通用底盘。在我们的项目中我们会实现一个简单的AgentHarness类负责启动TechnicalWritingAgent为其注入配置好的 Skills并处理执行过程中的异常。1.4 SubagentAgent 的“下属”与职责分解单元Subagent 是 Agent 概念的一个特化。当一个主 Agent 的任务过于复杂时它可以创建或委托任务给一个或多个 Subagent。Subagent 本身也是一个完整的 Agent拥有自己的目标、规划和能力Skills。主 Agent 和 Subagent 之间形成一种层次结构或协作网络。Subagent 的核心价值在于模块化和关注点分离。例如我们的TechnicalWritingAgent在规划“写正文”这个子任务时可能会觉得这部分逻辑也很复杂涉及引言、技术细节、代码示例、总结等多个部分。它可以创建一个DraftWritingSubagent并将“根据大纲和资料撰写正文”这个目标委托给它。DraftWritingSubagent内部可以再进行更细粒度的规划和 Skill 调用。这样主 Agent 的逻辑得以简化系统也更容易扩展和维护。它们之间的关系可以概括为Harness 提供舞台Agent和 Subagent是舞台上的演员Skill 是演员手中的道具。演员Agent根据剧本目标决定行动顺序并适时使用道具Skill或与其他演员Subagent配合来完成演出。2. 项目设计与环境准备我们将构建一个简化的“技术文章助手”系统来演示上述概念。该系统接收一个技术主题如“Python 装饰器”最终输出一篇 Markdown 格式的博客草稿。2.1 系统架构设计[用户输入] - [Agent Harness] - [TechnicalWritingAgent (主Agent)] | |-- 规划任务 | |-- 执行: 调用 WebSearchSkill |-- 执行: 调用 CodeAnalysisSkill (如有代码) |-- 执行: 创建 DraftWritingSubagent | |-- Subagent 调用 TextGenerationSkill |-- 执行: 调用 MarkdownFormatSkill | V [Markdown 输出]2.2 环境与依赖配置本项目使用 Python 作为实现语言因为它有丰富的 AI 相关库和清晰的语法。我们主要使用openai库来模拟 LLM 调用实际可用 GPT、Claude、DeepSeek 等 API并用requests模拟网络搜索。首先创建项目目录并初始化虚拟环境mkdir tech-blog-agent cd tech-blog-agent python -m venv venv # Windows venv\Scripts\activate # Linux/Mac source venv/bin/activate创建requirements.txt文件列出核心依赖openai1.0.0 requests2.28.0 pydantic2.0.0 python-dotenv1.0.0安装依赖pip install -r requirements.txt为了安全地管理 API 密钥等配置我们使用环境变量。创建.env文件记得加入.gitignoreOPENAI_API_KEYyour_openai_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 # 或你的代理地址 SEARCH_API_ENDPOINThttps://api.example.com/search # 模拟搜索端点创建config.py来读取配置import os from dotenv import load_dotenv load_dotenv() class Config: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) OPENAI_BASE_URL os.getenv(OPENAI_BASE_URL) SEARCH_API_ENDPOINT os.getenv(SEARCH_API_ENDPOINT) classmethod def validate(cls): if not cls.OPENAI_API_KEY: raise ValueError(OPENAI_API_KEY 未在环境变量中设置) # 其他必要配置的校验3. 实现核心组件从 Skill 到 Harness我们将采用自底向上的方式先实现最基础的 Skill再构建 Agent 和 Harness。3.1 实现基础 Skill 类首先定义一个所有 Skill 的基类用于规范接口。创建skills/base_skill.pyfrom abc import ABC, abstractmethod from typing import Any, Dict from pydantic import BaseModel class SkillInput(BaseModel): Skill 输入参数的基类使用 Pydantic 进行验证和序列化。 pass class SkillOutput(BaseModel): Skill 输出结果的基类。 success: bool data: Any error_message: str class BaseSkill(ABC): 所有 Skill 的抽象基类。 name: str base_skill description: str 一个基础的 Skill abstractmethod async def execute(self, input_data: SkillInput) - SkillOutput: 执行 Skill 的核心逻辑。 pass def to_dict(self) - Dict[str, str]: 返回 Skill 的描述信息供 Agent 发现和选择。 return {name: self.name, description: self.description}3.2 实现具体 Skills接下来实现几个具体的 Skill。创建skills/web_search_skill.pyimport requests from typing import List from pydantic import Field from skills.base_skill import BaseSkill, SkillInput, SkillOutput class WebSearchInput(SkillInput): query: str Field(..., description搜索查询词) max_results: int Field(5, description最大返回结果数) class WebSearchSkill(BaseSkill): name web_search description 根据查询词从网络获取相关信息摘要 def __init__(self, api_endpoint: str): self.api_endpoint api_endpoint async def execute(self, input_data: WebSearchInput) - SkillOutput: try: # 模拟调用搜索 API实际项目中替换为真实调用 # 这里使用 requests 进行同步调用在生产环境应考虑异步 HTTP 客户端如 httpx response requests.get( self.api_endpoint, params{q: input_data.query, limit: input_data.max_results}, timeout10 ) response.raise_for_status() # 假设 API 返回 JSON 格式{results: [{title: ..., snippet: ...}, ...]} search_results response.json().get(results, []) # 简化处理只提取摘要文本 summaries [f{res.get(title, )}: {res.get(snippet, )} for res in search_results] return SkillOutput(successTrue, datasummaries) except Exception as e: return SkillOutput(successFalse, dataNone, error_messagef搜索失败: {str(e)})创建skills/text_generation_skill.pyfrom openai import OpenAI from pydantic import Field from skills.base_skill import BaseSkill, SkillInput, SkillOutput from config import Config class TextGenerationInput(SkillInput): prompt: str Field(..., description生成文本的提示词) max_tokens: int Field(1000, description生成的最大 token 数) temperature: float Field(0.7, description生成温度控制随机性) class TextGenerationSkill(BaseSkill): name text_generation description 调用大语言模型生成文本内容 def __init__(self): self.client OpenAI(api_keyConfig.OPENAI_API_KEY, base_urlConfig.OPENAI_BASE_URL) async def execute(self, input_data: TextGenerationInput) - SkillOutput: try: response self.client.chat.completions.create( modelgpt-3.5-turbo, # 可根据需要调整模型 messages[{role: user, content: input_data.prompt}], max_tokensinput_data.max_tokens, temperatureinput_data.temperature ) generated_text response.choices[0].message.content return SkillOutput(successTrue, datagenerated_text) except Exception as e: return SkillOutput(successFalse, dataNone, error_messagef文本生成失败: {str(e)})类似地你可以实现CodeAnalysisSkill可能调用 AST 解析库和MarkdownFormatSkill使用正则表达式或markdown库进行格式化。为了简化示例我们假设它们已存在。3.3 实现 Agent HarnessHarness 是系统的粘合剂。创建harness/agent_harness.pyimport asyncio import logging from typing import Dict, List, Optional from skills.base_skill import BaseSkill class AgentHarness: Agent 运行框架管理 Skill 注册和提供基础服务。 def __init__(self, name: str default_harness): self.name name self.skills: Dict[str, BaseSkill] {} self.logger logging.getLogger(name) self._setup_logging() def _setup_logging(self): logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s ) def register_skill(self, skill: BaseSkill): 向 Harness 注册一个 Skill供 Agent 使用。 if skill.name in self.skills: self.logger.warning(fSkill {skill.name} 已存在将被覆盖。) self.skills[skill.name] skill self.logger.info(f已注册 Skill: {skill.name}) def get_skill(self, skill_name: str) - Optional[BaseSkill]: 根据名称获取已注册的 Skill。 return self.skills.get(skill_name) def list_skills(self) - List[Dict]: 列出所有可用的 Skill 及其描述。 return [skill.to_dict() for skill in self.skills.values()] async def run_agent(self, agent_instance): 运行一个 Agent 实例。这是 Harness 的核心管理方法。 self.logger.info(fHarness {self.name} 开始运行 Agent: {agent_instance.__class__.__name__}) try: # 这里可以添加前置钩子如状态加载、权限检查等 result await agent_instance.run() # 这里可以添加后置钩子如状态保存、结果上报等 self.logger.info(fAgent 执行完成。) return result except asyncio.CancelledError: self.logger.warning(Agent 任务被取消。) raise except Exception as e: self.logger.error(fAgent 运行过程中发生未捕获异常: {e}, exc_infoTrue) raise3.4 实现主 Agent 与 Subagent现在实现核心的决策者。创建agents/technical_writing_agent.pyimport asyncio from typing import List, Dict, Any from pydantic import BaseModel, Field from harness.agent_harness import AgentHarness from skills.base_skill import SkillInput class AgentContext(BaseModel): Agent 执行上下文存储任务状态和中间结果。 user_input: str search_results: List[str] Field(default_factorylist) outline: str draft: str final_output: str class TechnicalWritingAgent: 技术文章生成主 Agent。 def __init__(self, harness: AgentHarness, user_topic: str): self.harness harness self.context AgentContext(user_inputuser_topic) self.logger harness.logger.getChild(self.__class__.__name__) async def run(self) - str: 主 Agent 的执行入口定义了任务规划流程。 self.logger.info(f开始处理主题: {self.context.user_input}) # 1. 规划阶段定义任务步骤 plan [ self._step_search_web, self._step_generate_outline, self._step_write_draft, self._step_format_output ] # 2. 执行阶段按顺序执行每个步骤 for step_func in plan: step_name step_func.__name__ self.logger.info(f执行步骤: {step_name}) try: success await step_func() if not success: self.logger.error(f步骤 {step_name} 执行失败终止流程。) break except Exception as e: self.logger.error(f步骤 {step_name} 执行异常: {e}, exc_infoTrue) raise self.logger.info(f文章生成完成。) return self.context.final_output async def _step_search_web(self) - bool: 步骤1调用 WebSearchSkill 收集资料。 search_skill self.harness.get_skill(web_search) if not search_skill: self.logger.error(未找到 web_search skill) return False from skills.web_search_skill import WebSearchInput input_data WebSearchInput(queryf{self.context.user_input} 技术博客 最新, max_results3) result await search_skill.execute(input_data) if result.success: self.context.search_results result.data self.logger.info(f搜索完成获得 {len(result.data)} 条结果。) return True else: self.logger.error(f搜索失败: {result.error_message}) return False async def _step_generate_outline(self) - bool: 步骤2调用 TextGenerationSkill 生成文章大纲。 # 构建提示词利用上一步的搜索结果 search_context \n.join(self.context.search_results[:2]) # 取前两条作为上下文 prompt f 请作为一名资深技术博主为主题“{self.context.user_input}”撰写一篇技术博客大纲。 请参考以下背景信息 {search_context} 大纲要求 1. 包含引言、核心概念、实战示例、常见问题、总结等部分。 2. 每个部分列出2-3个关键子点。 3. 输出格式为清晰的列表。 text_gen_skill self.harness.get_skill(text_generation) if not text_gen_skill: self.logger.error(未找到 text_generation skill) return False from skills.text_generation_skill import TextGenerationInput input_data TextGenerationInput(promptprompt, max_tokens500) result await text_gen_skill.execute(input_data) if result.success: self.context.outline result.data self.logger.info(大纲生成完成。) return True else: self.logger.error(f大纲生成失败: {result.error_message}) return False async def _step_write_draft(self) - bool: 步骤3创建 Subagent 来负责撰写详细草稿。 self.logger.info(创建 DraftWritingSubagent 来撰写正文草稿...) # 实例化 Subagent subagent DraftWritingSubagent( harnessself.harness, topicself.context.user_input, outlineself.context.outline, referencesself.context.search_results ) # 运行 Subagent draft_result await subagent.run() if draft_result: self.context.draft draft_result self.logger.info(Subagent 草稿撰写完成。) return True else: self.logger.error(Subagent 草稿撰写失败。) return False async def _step_format_output(self) - bool: 步骤4调用 MarkdownFormatSkill 进行最终格式化。 format_skill self.harness.get_skill(markdown_format) if not format_skill: self.logger.warning(未找到 markdown_format skill使用原始草稿。) self.context.final_output self.context.draft return True # 假设 MarkdownFormatSkill 的输入需要一个 content 字段 from skills.markdown_format_skill import MarkdownFormatInput input_data MarkdownFormatInput(contentself.context.draft) result await format_skill.execute(input_data) if result.success: self.context.final_output result.data self.logger.info(Markdown 格式化完成。) return True else: self.logger.warning(f格式化失败使用未格式化的草稿: {result.error_message}) self.context.final_output self.context.draft return True接着在同一个文件中或新建agents/draft_writing_subagent.py实现 Subagentclass DraftWritingSubagent: 专门负责撰写文章正文的 Subagent。 def __init__(self, harness, topic: str, outline: str, references: List[str]): self.harness harness self.topic topic self.outline outline self.references references self.logger harness.logger.getChild(self.__class__.__name__) async def run(self) - str: Subagent 的执行逻辑根据大纲分段生成内容。 self.logger.info(fDraftWritingSubagent 开始工作主题: {self.topic}) # 简化的逻辑将大纲的每一部分作为提示词调用文本生成 Skill text_gen_skill self.harness.get_skill(text_generation) if not text_gen_skill: self.logger.error(Subagent 未找到 text_generation skill) return from skills.text_generation_skill import TextGenerationInput # 假设大纲是以“-”或“1.”开头的行 outline_lines [line.strip() for line in self.outline.split(\n) if line.strip().startswith((-, 1., 2., 3.))] full_draft_parts [] for line in outline_lines: prompt f 你正在撰写一篇关于“{self.topic}”的技术博客。 当前需要撰写的部分是{line} 以下是可供参考的资料摘要 {chr(10).join(self.references[:2])} 请围绕这个部分撰写详细、专业、可读性强的技术内容约300-500字。 input_data TextGenerationInput(promptprompt, max_tokens800) result await text_gen_skill.execute(input_data) if result.success: full_draft_parts.append(f## {line}\n\n{result.data}\n) else: self.logger.warning(f生成部分 {line} 时失败: {result.error_message}) full_draft_parts.append(f## {line}\n\n[此部分内容生成失败]\n) full_draft \n.join(full_draft_parts) self.logger.info(DraftWritingSubagent 工作完成。) return full_draft4. 组装与运行启动完整的 Agent 系统所有组件准备就绪后我们需要一个入口脚本将它们组装起来并运行。创建main.pyimport asyncio from config import Config from harness.agent_harness import AgentHarness from skills.web_search_skill import WebSearchSkill from skills.text_generation_skill import TextGenerationSkill from skills.markdown_format_skill import MarkdownFormatSkill # 假设已实现 from agents.technical_writing_agent import TechnicalWritingAgent async def main(): # 1. 验证配置 Config.validate() # 2. 初始化 Harness运行框架 harness AgentHarness(nameTechBlogHarness) # 3. 创建并注册 Skills工具 # 注意这里 SEARCH_API_ENDPOINT 是模拟的实际需要替换或使用 Mock search_skill WebSearchSkill(api_endpointConfig.SEARCH_API_ENDPOINT or https://httpbin.org/get) text_gen_skill TextGenerationSkill() format_skill MarkdownFormatSkill() # 假设已实现 harness.register_skill(search_skill) harness.register_skill(text_gen_skill) harness.register_skill(format_skill) print(已注册 Skills:, harness.list_skills()) # 4. 创建主 Agent并注入 Harness使其能访问 Skills user_topic input(请输入你想要生成博客的技术主题例如Python装饰器: ).strip() if not user_topic: user_topic Python异步编程 # 默认主题 agent TechnicalWritingAgent(harnessharness, user_topicuser_topic) # 5. 通过 Harness 运行 Agent print(f\n开始为主题『{user_topic}』生成技术博客...) try: final_article await harness.run_agent(agent) print(\n *50) print(生成的技术博客草稿) print(*50) print(final_article) print(*50) # 可选保存到文件 with open(f{user_topic}_blog_draft.md, w, encodingutf-8) as f: f.write(final_article) print(f\n草稿已保存至{user_topic}_blog_draft.md) except Exception as e: print(f\n系统运行出错: {e}) if __name__ __main__: asyncio.run(main())4.1 运行验证与结果分析在运行前请确保已正确设置OPENAI_API_KEY。由于SEARCH_API_ENDPOINT是模拟的WebSearchSkill可能会失败或返回模拟数据。为了演示我们可以先将其替换为一个返回固定数据的 Mock Skill或者使用一个真实的公开 API如 DuckDuckGo Instant Answer API 的模拟。运行程序python main.py输入一个技术主题例如“Python 列表推导式”。程序将按以下流程执行你可以在控制台看到对应的日志Harness 启动注册 Skills。主 Agent 开始运行。主 Agent 调用WebSearchSkill可能返回模拟数据。主 Agent 调用TextGenerationSkill生成大纲。主 Agent 创建DraftWritingSubagent。Subagent 多次调用TextGenerationSkill根据大纲各部分生成正文。主 Agent 调用MarkdownFormatSkill进行格式化。最终输出 Markdown 格式的草稿并保存到文件。一个成功的运行输出应该包含结构清晰的 Markdown 内容包含标题、章节和段落。5. 常见问题排查与调试技巧在实际集成和运行此类系统时你会遇到各种问题。以下是基于此项目的典型排查路径。5.1 Skill 执行失败问题现象可能原因检查点解决方案SkillNotFound错误1. Skill 未在 Harness 中注册。2. Skill 名称拼写错误。1. 检查main.py中register_skill调用。2. 检查 Agent 代码中get_skill的参数是否与注册名完全一致。1. 确保注册代码被执行。2. 使用harness.list_skills()打印所有已注册 Skill 进行核对。API 调用超时或网络错误1. 网络连接问题。2. API 端点错误或不可用。3. 未配置代理如需。1. 使用requests或curl手动测试 API 端点。2. 检查Config中的 URL 和密钥。3. 查看 Skill 的execute方法中的错误日志。1. 修复网络配置。2. 使用有效的 API 端点或 Mock 数据。3. 在 Skill 实现中增加更详细的错误处理和重试逻辑。输入参数验证失败1. 传递给 Skill 的input_data类型错误。2. Pydantic 模型字段缺失或类型不匹配。1. 检查调用 Skill 时传入的参数对象是否正确实例化。2. 查看 Pydantic 报错信息确认缺失字段。1. 确保使用正确的SkillInput子类创建输入对象。2. 为可选字段设置合理的默认值。5.2 Agent 逻辑问题问题现象可能原因检查点解决方案Agent 流程卡住或提前结束1. 某个步骤的success判断逻辑有误。2. 异步任务被意外取消。3. 异常未被捕获导致流程中断。1. 检查每个_step_*方法的返回值逻辑。2. 查看 Harness 和 Agent 的日志输出。3. 在run_agent和run方法中添加更全面的异常捕获。1. 确保每个步骤在失败时返回False并由主流程处理。2. 使用asyncio的调试工具或添加更多logger.info来跟踪执行流。Subagent 未按预期工作1. Subagent 未接收到正确的上下文数据。2. Subagent 内部的 Skill 调用失败。3. 主 Agent 和 Subagent 之间的职责划分不清。1. 打印 Subagent 初始化时接收到的topic,outline等数据。2. 检查 Subagent 内部是否能够正确获取到所需的 Skill。1. 确保主 Agent 传递给 Subagent 的数据是完整和准确的。2. 考虑让 Subagent 也接收harness实例以便共享 Skill 池。3. 重新审视任务分解的粒度确保 Subagent 的目标足够单一。5.3 配置与依赖问题问题现象可能原因检查点解决方案导入模块失败 (ModuleNotFoundError)1. 文件路径或模块名错误。2.__init__.py文件缺失导致包识别失败。3. PYTHONPATH 未包含项目根目录。1. 检查from ... import ...语句的路径。2. 确保每个子目录如skills/,agents/下存在__init__.py文件可以是空文件。3. 在 IDE 中或将项目根目录添加到系统路径。1. 使用相对导入或绝对导入时保持一致性。2. 创建缺失的__init__.py文件。3. 在main.py开头使用sys.path.append添加项目根目录仅用于开发调试。API 密钥无效1..env文件未加载。2. 环境变量名错误。3. API 密钥已过期或被禁用。1. 检查python-dotenv的load_dotenv()是否在Config类之前调用。2. 使用print(os.getenv(“OPENAI_API_KEY”))调试。3. 在对应 API 提供商的控制台检查密钥状态。1. 确保.env文件在正确的工作目录下。2. 核对.env文件中的变量名与config.py中读取的名称完全一致。3. 申请新的 API 密钥。调试建议在开发初期可以暂时将可能失败的 Skill如WebSearchSkill替换为一个返回固定数据的MockSkill以便快速验证主流程和 Agent 的逻辑是否正确。6. 生产环境最佳实践与扩展方向上述示例是一个用于理解概念的简化版本。要将此类系统用于生产需要考虑更多工程化因素。6.1 架构与性能优化异步与非阻塞所有 I/O 密集型操作网络请求、数据库查询、文件读写都应使用异步模式。示例中使用了async/await但requests库是同步的。生产环境应替换为httpx或aiohttp等异步 HTTP 客户端。Skill 池与缓存频繁创建的 Skill 实例尤其是包含重型客户端如 LLM Client 的应考虑复用或使用连接池。对于耗时的 Skill 调用结果可以引入缓存机制如redis并设置合理的过期时间。Harness 作为服务在微服务架构中Harness 可以独立部署为一个“Agent 运行时服务”通过 RPC 或消息队列接收任务管理多个 Agent 实例的生命周期并提供统一的监控、日志和调度能力。Subagent 的并发执行如果多个 Subagent 任务间没有依赖可以利用asyncio.gather并发执行提升整体效率。6.2 可观测性与稳定性结构化日志使用structlog或logging的 JSON Formatter输出包含agent_id,skill_name,request_id,duration等字段的结构化日志便于后续使用 ELK 或 Loki 进行聚合分析。分布式追踪为每个用户请求生成唯一的trace_id并在 Harness、Agent、Subagent、Skill 之间传递。集成 OpenTelemetry 等工具可视化整个调用链路的耗时和状态。熔断与降级对于依赖外部 API 的 Skill如 LLM、搜索实现熔断器模式如pybreaker。当外部服务不稳定时快速失败或切换到降级方案如返回缓存数据、使用更简单的本地模型。配置管理将 Skill 的参数如 API 端点、超时时间、重试次数从代码中抽离使用配置中心如 Apollo, Nacos或环境变量管理支持动态更新。6.3 扩展系统能力动态 Skill 发现当前的 Skill 注册是静态的。可以设计一个 Skill 注册中心支持热插拔。新的 Skill 服务启动后自动向注册中心上报Harness 定期拉取或订阅更新使 Agent 能动态发现新能力。Agent 记忆与学习为 Agent 引入持久化记忆层如向量数据库存储历史对话和任务结果。在规划新任务时Agent 可以先从记忆中进行检索获得相关上下文实现更连贯的交互和持续学习。多模态 Skill除了文本可以扩展处理图像、音频的 Skill。例如ImageAnalysisSkill、SpeechToTextSkill使 Agent 能够处理更复杂的多模态任务。工作流引擎集成对于极其复杂、流程固定的任务可以将 Agent 的规划逻辑卸载到专门的工作流引擎如 Airflow, Temporal。Harness 负责执行工作流引擎编排好的具体 Skill 调用步骤。通过这个从零搭建的“技术文章助手”项目我们清晰地拆解了 Skill、Agent、Harness、Subagent 的角色与协作方式。Skill 是工具Agent 是使用工具并做决策的“大脑”Harness 是提供支持和管理的“舞台”Subagent 则是“大脑”为了处理复杂子任务而创造的“分身”。理解这个模型后你可以将其应用于客服机器人、智能数据分析、自动化运维等众多场景。下一步尝试为你的 Agent 增加一个SQLQuerySkill让它能够从数据库获取数据或者实现一个HumanFeedbackSkill在关键决策点引入人工确认这将使你的 Agent 系统更加实用和强大。