使用Oh My Subagents框架构建本地多智能体AI团队：从原理到实践

在实际 AI 应用开发中，构建一个能够协同工作的智能体（Agent）团队正成为解决复杂任务的关键模式。单个智能体往往能力有限，而将多个具备不同专长的智能体组织起来，通过分工协作来处理一个流程，可以显著提升系统的整体能力和可靠性。然而，从零开始搭建这样的多智能体系统，涉及到智能体定义、通信机制、任务编排、状态管理等一系列复杂问题，开发门槛较高。

Oh My Subagents 是一个旨在简化这一过程的开源框架。它提供了一个轻量级的本地运行时环境，让开发者能够以声明式的方式快速定义和运行由多个子智能体（Subagents）组成的 AI 团队。其核心思想是将复杂的 AI 工作流拆解为一系列职责明确的子智能体，并通过清晰的接口和消息传递机制将它们连接起来，从而构建出结构清晰、易于理解和维护的多智能体应用。

本文面向对 AI 应用开发、智能体架构感兴趣的开发者。我们将从零开始，带你理解 Oh My Subagents 的核心概念，完成本地环境的搭建，并通过一个完整的示例项目来演示如何定义子智能体、编排任务流程以及运行和调试整个 AI 团队。你将学习到如何利用这个框架来组织你的 AI 逻辑，使其更具模块化和可扩展性。

1. 理解 Oh My Subagents 的核心概念与工作机制

在深入代码之前，我们需要先厘清几个核心概念，这有助于理解框架的设计哲学和后续的配置、编码工作。

1.1 什么是 Subagent（子智能体）

在 Oh My Subagents 的语境中，一个 Subagent 并非一个完全独立、拥有长期记忆和复杂规划能力的通用 AI 智能体。相反，它被设计为一个功能单一、职责明确的处理单元。你可以将其类比为微服务架构中的一个服务，或者函数式编程中的一个纯函数（尽管它内部可能调用大语言模型）。每个子智能体通常负责一项具体的任务，例如：

• 文本分析子智能体：负责提取用户查询中的关键信息。
• 代码生成子智能体：根据分析结果生成特定编程语言的代码片段。
• 代码审查子智能体：检查生成的代码是否存在语法错误或潜在问题。
• 结果格式化子智能体：将最终结果整理成用户友好的格式（如 Markdown）。

这种设计遵循了“单一职责原则”，使得每个子智能体易于开发、测试和复用。

1.2 AI Team（AI 团队）与工作流

多个子智能体按照一定的逻辑顺序组合起来，就形成了一个AI Team。这个团队共同协作来完成一个更大的目标。团队内部的工作流定义了子智能体之间的协作方式，目前主要支持**顺序管道（Sequential Pipeline）**模式：

1. 用户输入或上一个子智能体的输出，作为当前子智能体的输入。
2. 当前子智能体处理输入，并产生输出。
3. 该输出被传递给下一个子智能体作为其输入。
4. 如此依次执行，直到最后一个子智能体产生最终输出。

这种管道模式清晰直观，适用于大多数具有明确步骤的任务，例如“分析 -> 生成 -> 审查 -> 交付”这样的流程。

1.3 本地运行时（Local Runtime）的优势

Oh My Subagents 强调“本地运行时”，这意味着整个 AI 团队的编排和执行发生在你的开发机器或服务器上，而不是依赖某个特定的云端服务平台。这带来了几个关键优势：

• 隐私与安全：敏感数据和业务逻辑无需离开本地环境。
• 成本可控：你可以自由选择后端的大语言模型（LLM）API（如 OpenAI, Anthropic，或本地部署的模型），并直接管理其调用成本。
• 开发调试友好：所有日志、中间状态都在本地，便于使用熟悉的工具（如 IDE 调试器、日志文件）进行排查。
• 无供应商锁定：框架专注于编排逻辑，与具体的 LLM 提供商解耦，迁移成本低。

1.4 消息（Message）与上下文（Context）

子智能体之间通过消息进行通信。一条消息通常包含role（如user,assistant,system）和content。框架会管理一个贯穿整个工作流的上下文，它包含了初始输入、每个子智能体产生的消息以及一些元数据。子智能体可以从上下文中读取之前步骤的信息，并将自己的输出追加到上下文中，供后续步骤使用。理解数据如何在上下文中流动是调试多智能体系统的关键。

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

要开始使用 Oh My Subagents，你需要准备基础的 Python 开发环境，并安装必要的依赖。

2.1 基础环境要求

确保你的系统满足以下条件：

• Python: 版本 3.8 或更高。这是运行框架的基础。
• 包管理工具:pip的最新版本。
• 代码编辑器: 如 VS Code, PyCharm 等。
• （可选）虚拟环境管理工具: 如venv,conda或poetry。强烈建议使用虚拟环境来隔离项目依赖。

你可以通过以下命令检查 Python 环境：

python --version pip --version

2.2 创建项目并安装 Oh My Subagents

首先，创建一个新的项目目录并进入。

mkdir my-ai-team-project cd my-ai-team-project

接下来，创建一个虚拟环境（以venv为例）并激活它。

# 创建虚拟环境 python -m venv venv # 激活虚拟环境 # 在 Windows 上： venv\Scripts\activate # 在 macOS/Linux 上： source venv/bin/activate

激活后，命令行提示符前通常会显示(venv)，表示你已处于虚拟环境中。

现在，安装 Oh My Subagents 框架。由于它是一个较新的项目，我们通常直接从其源代码仓库安装。假设其 PyPI 包名为oh-my-subagents（请根据项目实际名称调整，这里作为示例），你可以尝试：

pip install oh-my-subagents

如果上述包名不可用，说明项目可能尚未发布到 PyPI，你需要从其 Git 仓库安装。这需要你先找到项目的仓库 URL（例如https://github.com/username/oh-my-subagents）。

pip install git+https://github.com/username/oh-my-subagents.git

安装完成后，验证安装是否成功。

python -c “import oh_my_subagents; print(oh_my_subagents.__version__)”

如果输出版本号或没有报错，说明安装成功。

2.3 配置 LLM 提供商 API 密钥

Oh My Subagents 本身不提供 AI 能力，它需要连接后端的大语言模型。这里以 OpenAI 的 GPT 模型为例。你需要在 OpenAI 官网注册并获取 API 密钥。

安全提醒：永远不要将 API 密钥硬编码在代码中或提交到版本控制系统。

推荐的做法是使用环境变量来管理密钥。在项目根目录创建一个名为.env的文件（确保该文件已被添加到.gitignore中），并写入你的密钥：

# .env 文件内容 OPENAI_API_KEY=sk-your-actual-openai-api-key-here

然后，在你的 Python 代码中，使用os模块或python-dotenv库来读取这个环境变量。首先安装python-dotenv：

pip install python-dotenv

3. 构建你的第一个 AI 团队：代码生成与审查助手

我们将构建一个简单的 AI 团队，它包含两个子智能体：一个负责根据自然语言描述生成 Python 代码，另一个负责审查生成的代码并提出改进建议。

3.1 项目结构设计

一个清晰的项目结构有助于管理代码。建议如下：

my-ai-team-project/ ├── .env # 环境变量（API密钥等） ├── .gitignore # Git忽略文件 ├── requirements.txt # 项目依赖列表 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── agents/ # 子智能体定义目录 │ │ ├── __init__.py │ │ ├── code_generator.py │ │ └── code_reviewer.py │ ├── team_definitions.py # AI团队定义 │ └── main.py # 应用入口 └── tests/ # 测试目录

使用pip freeze > requirements.txt生成依赖文件。

3.2 定义子智能体：代码生成器（CodeGenerator）

在src/agents/code_generator.py中，我们定义第一个子智能体。它的职责是接收用户的自然语言需求，并生成相应的 Python 代码。

# src/agents/code_generator.py import os from dotenv import load_dotenv from openai import OpenAI # 假设使用OpenAI客户端 from oh_my_subagents import Subagent # 导入框架基类 # 加载环境变量 load_dotenv() class CodeGenerator(Subagent): """子智能体：根据自然语言描述生成Python代码。""" def __init__(self, name="code_generator"): super().__init__(name=name) # 初始化OpenAI客户端，从环境变量读取API密钥 self.client = OpenAI(api_key=os.getenv(“OPENAI_API_KEY”)) # 定义系统提示词，设定该智能体的角色和能力 self.system_prompt = “””你是一个专业的Python开发助手。你的任务是根据用户的需求，生成正确、高效、可读性好的Python代码。 只返回代码本身，除非用户特别要求，否则不要包含任何解释性文字。 如果需求不明确，你可以询问，但在这个任务中请基于已有信息生成最佳代码。””” async def execute(self, context): """ 执行智能体的核心逻辑。 Args: context: 工作流上下文，包含之前的消息和历史。 Returns: 更新后的上下文。 """ # 从上下文中获取最新的用户输入。 # 假设用户输入是上下文中的最后一条‘user’消息。 user_messages = [msg for msg in context.messages if msg[“role”] == “user”] if not user_messages: # 如果没有用户输入，则使用一个默认任务或返回错误 user_input = “请生成一个Python函数，计算斐波那契数列的第n项。” else: user_input = user_messages[-1][“content”] # 调用大语言模型生成代码 try: response = self.client.chat.completions.create( model=“gpt-4o-mini”, # 可根据需要选择模型，如 gpt-4, gpt-3.5-turbo messages=[ {“role”: “system”, “content”: self.system_prompt}, {“role”: “user”, “content”: f”请生成Python代码：{user_input}”} ], temperature=0.2, # 低温度使输出更确定，适合代码生成 max_tokens=1000 ) generated_code = response.choices[0].message.content # 清理响应，确保我们只获取代码块（去除可能存在的markdown代码标记） if “`” in generated_code: # 简单处理，提取反引号内的内容 import re code_blocks = re.findall(r’`(.*?)`’, generated_code, re.DOTALL) if code_blocks: generated_code = code_blocks[-1] # 取最后一个代码块 else: generated_code = generated_code.strip(‘` \n’) except Exception as e: generated_code = f”# 代码生成失败: {str(e)}” # 将本智能体的输出添加到上下文中 context.add_message({ “role”: “assistant”, “name”: self.name, “content”: generated_code }) # 也可以将生成的代码存储到上下文的自定义数据中，便于后续智能体使用 context.set(“generated_code”, generated_code) return context

关键点解释：

1. 继承Subagent基类：这是定义子智能体的标准方式。
2. __init__方法：用于初始化智能体名称、客户端和系统提示词。系统提示词至关重要，它决定了智能体的行为边界。
3. execute方法：这是智能体的核心入口。它必须是async（异步）方法，接收并返回context对象。框架会负责调用它。
4. 上下文操作：context.messages存储了对话历史。context.add_message用于追加本智能体的输出。context.set/get用于在上下文间传递结构化数据，这比仅用消息更灵活。
5. 错误处理：在调用外部 API 时，务必进行异常捕获，避免单个智能体失败导致整个工作流崩溃。

3.3 定义子智能体：代码审查员（CodeReviewer）

在src/agents/code_reviewer.py中，我们定义第二个子智能体。它的职责是审查CodeGenerator生成的代码。

# src/agents/code_reviewer.py import os from dotenv import load_dotenv from openai import OpenAI from oh_my_subagents import Subagent load_dotenv() class CodeReviewer(Subagent): """子智能体：审查Python代码，提出改进建议。""" def __init__(self, name=“code_reviewer”): super().__init__(name=name) self.client = OpenAI(api_key=os.getenv(“OPENAI_API_KEY”)) self.system_prompt = “””你是一个资深的Python代码审查专家。你的任务是仔细检查提供的Python代码，从以下方面进行评估： 1. **正确性**：代码逻辑是否正确？是否存在语法错误或运行时错误？ 2. **效率**：算法复杂度是否最优？有无不必要的循环或重复计算？ 3. **可读性**：变量命名是否清晰？代码结构是否良好？注释是否恰当？ 4. **Python风格**：是否符合PEP 8规范？ 5. **健壮性**：是否考虑了边界条件（如空输入、非法输入）？ 请以清晰、有条理的方式列出发现的问题和改进建议。对于每个问题，请说明原因并提供修改后的代码片段（如果适用）。 最终给出一个总体评价（例如：优秀/良好/需要改进）。””” async def execute(self, context): # 从上下文中获取前一个智能体（CodeGenerator）生成的代码 # 方式一：从上下文的自定义数据中获取 code_to_review = context.get(“generated_code”) # 方式二：从消息历史中获取（如果CodeGenerator只通过消息传递） if not code_to_review: # 尝试从最新的‘assistant’消息中获取 assistant_messages = [msg for msg in context.messages if msg[“role”] == “assistant” and msg.get(“name”) == “code_generator”] if assistant_messages: code_to_review = assistant_messages[-1][“content”] if not code_to_review: review_result = “错误：未找到需要审查的代码。” else: try: response = self.client.chat.completions.create( model=“gpt-4o-mini”, messages=[ {“role”: “system”, “content”: self.system_prompt}, {“role”: “user”, “content”: f”请审查以下Python代码：\n\n{code_to_review}”} ], temperature=0.1, # 审查需要非常确定性的输出 max_tokens=1500 ) review_result = response.choices[0].message.content except Exception as e: review_result = f”代码审查失败: {str(e)}” # 将审查结果添加到上下文 context.add_message({ “role”: “assistant”, “name”: self.name, “content”: review_result }) context.set(“code_review”, review_result) return context

关键点解释：

1. 数据获取：CodeReviewer需要获取CodeGenerator的输出。这里演示了两种方式：通过context.get(“generated_code”)获取自定义数据，或从消息历史中过滤。前者更直接可靠，是推荐做法。
2. 系统提示词：审查智能体的提示词需要更详细，明确审查的维度和输出格式要求。
3. 错误处理：同样需要处理code_to_review为空和 API 调用异常的情况。

3.4 组装 AI 团队并定义工作流

在src/team_definitions.py中，我们将两个子智能体组装成一个团队，并定义它们的执行顺序。

# src/team_definitions.py from oh_my_subagents import Team from src.agents.code_generator import CodeGenerator from src.agents.code_reviewer import CodeReviewer def create_code_gen_review_team(): """创建并返回一个代码生成与审查团队。""" # 1. 实例化子智能体 generator = CodeGenerator() reviewer = CodeReviewer() # 2. 创建团队，并指定子智能体列表（顺序即执行顺序） team = Team( name=“code_generation_and_review_team”, subagents=[generator, reviewer] # 先执行generator，再执行reviewer ) return team

关键点解释：

1. Team类：这是框架中用于管理子智能体集合和执行流程的核心类。
2. 执行顺序：subagents列表的顺序决定了工作流的执行顺序。这是一个简单的顺序管道。

3.5 编写主程序并运行

最后，在src/main.py中，我们编写入口程序来使用这个 AI 团队。

# src/main.py import asyncio import sys from src.team_definitions import create_code_gen_review_team async def main(): # 1. 获取用户输入（从命令行参数或直接指定） if len(sys.argv) > 1: user_request = “ “.join(sys.argv[1:]) else: # 默认示例请求 user_request = “写一个函数，它接收一个整数列表，返回列表中所有偶数的平方和。” print(f“用户请求: {user_request}”) print(“-” * 50) # 2. 创建 AI 团队 team = create_code_gen_review_team() # 3. 创建初始上下文，并传入用户请求 from oh_my_subagents import Context initial_context = Context() initial_context.add_message({ “role”: “user”, “content”: user_request }) # 4. 运行团队！ print(“AI 团队开始执行...\n”) try: final_context = await team.run(initial_context) except Exception as e: print(f”团队执行过程中发生错误: {e}”) return # 5. 输出最终结果 print(“\n” + “=”*50) print(“最终输出:”) print(“=”*50) # 打印所有消息 for msg in final_context.messages: if msg[“role”] in [“user”, “system”]: continue # 跳过用户和系统初始消息，只看智能体输出 print(f”\n[{msg.get(‘name’, ‘unknown’)}]:”) print(msg[“content”]) print(“-” * 30) # 也可以从上下文中获取结构化数据 generated_code = final_context.get(“generated_code”) code_review = final_context.get(“code_review”) if generated_code: print(f”\n生成的代码（从上下文获取）:\n{generated_code}”) if code_review: print(f”\n代码审查意见（从上下文获取）:\n{code_review}”) if __name__ == “__main__”: asyncio.run(main())

4. 运行验证与结果分析

现在，让我们运行这个 AI 团队，看看它如何工作。

4.1 执行程序

在项目根目录下，运行以下命令：

python src/main.py “编写一个Python函数，判断一个字符串是否是回文。”

或者使用默认的示例请求：

python src/main.py

4.2 预期输出与解读

程序运行后，你会在控制台看到类似以下的输出（具体内容因模型和请求而异）：

用户请求: 编写一个Python函数，判断一个字符串是否是回文。 -------------------------------------------------- AI 团队开始执行... ================================================== 最终输出: ================================================== [code_generator]: def is_palindrome(s: str) -> bool: “”” 判断字符串是否为回文。 忽略大小写和非字母数字字符。 “”” # 清理字符串：转小写，移除非字母数字字符 cleaned = “.join(ch.lower() for ch in s if ch.isalnum()) # 比较清理后的字符串与其反转是否相等 return cleaned == cleaned[::-1] ------------------------------ [code_reviewer]: # 代码审查报告 ## 总体评价：良好 ## 详细审查： 1. **正确性**：逻辑正确。通过清理字符串后比较其反转，是判断回文的经典方法。 2. **效率**：O(n)时间复杂度，需要遍历字符串一次进行清理，再比较反转（也是O(n)）。对于此问题已是最优。 3. **可读性**： * 函数名 `is_palindrome` 清晰。 * 有文档字符串说明功能。 * 变量名 `cleaned` 易懂。 * 注释说明了关键步骤。 4. **Python风格**：符合 PEP 8。使用了类型提示 (`s: str` -> `bool`)。 5. **健壮性**：考虑了大小写和非字母数字字符，这是很好的实践。 ## 改进建议： * **（可选）性能微调**：如果追求极致性能，可以尝试在单次循环中同时从两端向中间比较原始字符串（跳过非字母数字字符），避免创建新的字符串。但对于大多数用例，当前实现已足够好且更清晰。 * **（可选）输入验证**：如果函数是公共API的一部分，可以考虑对输入 `s` 进行类型检查（例如 `if not isinstance(s, str): raise TypeError(...)`），但鉴于有类型提示，通常可以省略。 **修改后的代码（仅展示可选优化版本，原代码已很好）：** ```python def is_palindrome_optimized(s: str) -> bool: if not isinstance(s, str): raise TypeError(“Input must be a string”) left, right = 0, len(s) - 1 while left < right: # 跳过非字母数字字符 while left < right and not s[left].isalnum(): left += 1 while left < right and not s[right].isalnum(): right -= 1 if s[left].lower() != s[right].lower(): return False left += 1 right -= 1 return True

**输出分析：** 1. **流程清晰**：输出明确展示了 `code_generator` 和 `code_reviewer` 两个子智能体的依次执行。 2. **结果分离**：每个智能体的输出被清晰地标记和分隔。 3. **内容质量**： * `code_generator` 生成了功能正确、带有文档字符串和类型提示的 Python 代码。 * `code_reviewer` 给出了结构化的审查报告，包含总体评价、多维度分析和具体的改进建议（甚至提供了另一种实现）。 4. **上下文传递成功**：`code_reviewer` 成功获取到了 `code_generator` 生成的代码，证明团队内的数据流转是正常的。 ### 4.3 验证方式与检查点 在运行后，你应该检查以下几点以验证团队是否正常工作： | 检查点 | 验证方法 | 预期结果 | | :--- | :--- | :--- | | **团队能否启动** | 观察控制台初始输出，是否出现导入错误或初始化错误。 | 程序正常启动，打印“用户请求”和“AI 团队开始执行”。 | | **子智能体顺序执行** | 观察输出中智能体名称出现的顺序。 | 先出现 `[code_generator]:`，后出现 `[code_reviewer]:`。 | | **数据传递正确** | 检查 `code_reviewer` 的输出是否针对 `code_generator` 生成的代码进行评论。 | 审查报告中的代码示例应与生成的代码功能一致。 | | **API 调用成功** | 观察是否有网络超时、认证失败或额度不足的错误信息。 | 输出为正常的代码和文本，而非错误堆栈。 | | **最终上下文完整** | 在 `main.py` 中打印 `final_context.messages` 的长度和内容。 | `messages` 列表应包含 user, code_generator, code_reviewer 的消息。 | ## 5. 常见问题排查与调试技巧 在开发和使用 Oh My Subagents 团队时，你可能会遇到一些问题。以下是常见问题的排查路径。 ### 5.1 问题：导入错误（ModuleNotFoundError） * **现象**：运行 `python src/main.py` 时，报错 `ModuleNotFoundError: No module named ‘oh_my_subagents’` 或 `No module named ‘src’`。 * **可能原因 1**：Oh My Subagents 包未正确安装。 * **检查**：在激活的虚拟环境中运行 `pip list | grep subagents`（或 `pip list` 后手动查找）。 * **解决**：重新执行安装命令 `pip install oh-my-subagents` 或 Git 安装命令。 * **可能原因 2**：Python 解释器路径错误，未使用虚拟环境中的 Python。 * **检查**：在终端输入 `which python`（或 `where python` on Windows），确认路径指向项目 `venv` 目录下。 * **解决**：确保虚拟环境已激活，或在 IDE 中正确配置了解释器。 * **可能原因 3**：`src` 目录未被识别为模块。 * **检查**：确保 `src` 目录下存在 `__init__.py` 文件（即使是空的）。 * **解决**：创建该文件。 ### 5.2 问题：API 密钥错误或网络问题 * **现象**：程序在调用 `client.chat.completions.create` 时卡住、超时或返回认证错误。 * **可能原因 1**：环境变量未正确加载。 * **检查**：在 `main.py` 开头临时添加 `print(os.getenv(“OPENAI_API_KEY”))`，查看是否打印出密钥（前几位和后几位）。 * **解决**：确认 `.env` 文件在项目根目录，且内容格式正确（无多余空格）。确保 `python-dotenv` 已安装，且 `load_dotenv()` 在创建客户端之前被调用。 * **可能原因 2**：网络连接问题或 API 服务不可用。 * **检查**：尝试在命令行用 `curl` 或使用其他工具测试 OpenAI API 连通性。检查防火墙或代理设置。 * **解决**：配置网络或等待服务恢复。 * **可能原因 3**：API 额度不足或模型不可用。 * **检查**：登录 OpenAI 控制台查看额度和账单。 * **解决**：充值或更换为有额度的 API 密钥，或检查模型名称是否正确。 ### 5.3 问题：子智能体未按预期顺序执行或未执行 * **现象**：只有第一个智能体有输出，或者输出顺序混乱。 * **可能原因 1**：`Team` 的 `subagents` 列表顺序错误。 * **检查**：查看 `team_definitions.py` 中 `Team` 实例化时的列表顺序。 * **解决**：调整列表顺序。 * **可能原因 2**：某个子智能体的 `execute` 方法出现未处理的异常，导致工作流中断。 * **检查**：在每个子智能体的 `execute` 方法内部添加更详细的 `try-except` 块，打印或记录错误信息。 * **解决**：修复导致异常的代码（如数据格式错误、API 调用参数错误）。 * **可能原因 3**：`execute` 方法不是 `async` 异步方法，或没有正确返回 `context`。 * **检查**：确认每个子智能体类中的 `execute` 方法定义包含 `async` 关键字，并且最后返回了 `context` 对象。 * **解决**：修正方法签名和返回值。 ### 5.4 问题：上下文（Context）数据传递失败 * **现象**：后一个智能体获取不到前一个智能体设置的数据（`context.get` 返回 `None`）。 * **可能原因 1**：使用的键（Key）不一致。 * **检查**：前一个智能体使用 `context.set(“my_key”, data)`，后一个智能体必须使用完全相同的字符串 `“my_key”` 来 `get`。 * **解决**：将键名定义为常量，或在团队内部约定好键名。 * **可能原因 2**：`context` 对象在传递过程中被意外替换或修改。 * **检查**：确保每个智能体的 `execute` 方法接收的是同一个 `context` 对象，并且返回的是更新后的同一个对象（通常是 `return context`）。 * **解决**：不要在 `execute` 方法内部创建新的 `Context` 实例。 ### 5.5 调试技巧 1. **增加日志**：在子智能体的 `execute` 方法开始和结束时打印日志，记录输入和输出。 ```python import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) async def execute(self, context): logger.info(f”{self.name} 开始执行，输入消息数: {len(context.messages)}”) # ... 处理逻辑 ... logger.info(f”{self.name} 执行完毕，输出: {output[:100]}...”) # 截断长输出 return context ``` 2. **检查中间状态**：在 `main.py` 的 `team.run` 调用前后，打印 `context` 的详细信息。 3. **简化测试**：暂时注释掉部分子智能体，或者用简单的模拟逻辑（如直接返回固定字符串）替换复杂的 LLM 调用，先确保框架流程正确。 4. **使用同步模式（如果支持）**：如果框架支持同步运行，可以先使用同步方式调试，避免异步带来的复杂性。 ## 6. 最佳实践与扩展方向 掌握了基础用法后，遵循一些最佳实践能让你的 AI 团队更健壮、更易维护。 ### 6.1 子智能体设计最佳实践 * **职责单一**：这是最重要的原则。一个子智能体只做一件事，并把它做好。这有利于测试、复用和组合。 * **明确的输入输出契约**：在智能体的文档或系统提示词中，清晰定义它期望的输入格式和产生的输出格式。例如，“本智能体输入应为一段文本，输出为该文本的情感分析结果（积极/消极/中性）”。 * **健壮的错误处理**：`execute` 方法内部必须包含 `try-except`，处理可能出现的所有异常（网络、解析、业务逻辑错误），并返回一个包含错误信息的上下文，而不是让异常抛出导致整个工作流崩溃。 * **可配置化**：将模型名称、温度（temperature）、最大令牌数（max_tokens）等参数作为智能体构造函数的参数，而不是硬编码在内部。这样可以在不同场景下复用同一个智能体。 * **状态无状态化**：尽量将子智能体设计为无状态的。执行逻辑只依赖于输入的 `context`，不依赖类内部的可变成员变量（除非是缓存等优化）。这保证了智能体在并发环境下的安全性。 ### 6.2 团队编排与上下文管理最佳实践 * **使用上下文存储结构化数据**：对于复杂的数据（如字典、列表、对象），优先使用 `context.set(key, value)` 和 `context.get(key)` 来传递，而不是仅仅依赖非结构化的消息历史。这使数据获取更准确、高效。 * **定义团队级别的输入输出规范**：为整个团队定义清晰的输入格式和最终输出格式。例如，团队输入必须是一个包含 `task_type` 和 `description` 字段的 JSON 字符串，最终输出必须包含 `code` 和 `review_report` 字段。 * **实现条件分支与循环**：基础的顺序管道可能不够。研究框架是否支持更复杂的工作流模式（如基于上下文内容的条件路由、循环执行某个智能体直到满足条件）。如果原生不支持，可以考虑在单个“协调者”智能体内实现简单逻辑，或寻找支持 DAG（有向无环图）的扩展。 * **添加超时与重试机制**：对于调用外部 API 的子智能体，实现超时控制。对于可重试的错误（如网络抖动），加入有限次数的重试逻辑。 ### 6.3 生产环境考量 * **配置外部化**：将所有配置（API 端点、密钥、模型参数、超时时间）移到配置文件（如 `config.yaml`）或环境变量中，与代码分离。 * **日志与监控**：集成成熟的日志库（如 `structlog`），为每个子智能体的执行记录详细的结构化日志，包括开始/结束时间、输入/输出摘要、错误信息等。这便于问题追踪和性能分析。 * **性能优化**： * **缓存**：对于计算成本高或结果相对稳定的子智能体（如文本向量化），考虑引入缓存机制。 * **并发执行**：如果子智能体之间没有依赖关系，探索框架是否支持并发执行（例如 `asyncio.gather`），以缩短整体延迟。 * **版本管理**：对子智能体的定义、团队的组合方式、使用的提示词进行版本控制。当 AI 团队的行为发生变化时，能够清晰地追溯和回滚。 ### 6.4 扩展方向 * **集成更多工具**：让子智能体不仅能调用 LLM，还能调用外部工具，如数据库查询、计算器、搜索引擎 API、内部系统接口等。这可以极大扩展 AI 团队的能力边界。 * **实现记忆与持久化**：将重要的上下文或对话历史持久化到数据库，使 AI 团队能够在多次交互中保持“记忆”，实现更复杂的多轮对话任务。 * **构建可视化编排界面**：对于复杂的团队，一个图形化的拖拽式编排界面可以极大提升开发效率。你可以考虑基于 Oh My Subagents 的核心 API 来构建这样的前端。 * **探索更复杂的团队拓扑**：超越简单的管道，尝试树状、图状的工作流。例如，一个“决策”智能体根据输入分析结果，将任务分发给不同的“专家”智能体并行处理，最后再由一个“汇总”智能体整合结果。 通过 Oh My Subagents 框架，你将复杂的 AI 应用逻辑分解为一个个可管理、可测试的子模块，并通过清晰的接口将它们组装起来。这种模式不仅使代码更易于维护，也让你能更灵活地试验不同的智能体组合和提示词策略，从而迭代出更强大的 AI 解决方案。从今天构建的简单代码审查助手开始，尝试加入更多的子智能体（如单元测试生成器、文档编写器），探索更丰富的应用场景。