AI参谋长代理:使用Claude Code SDK构建全面代理的指南
技术背景
随着大语言模型能力的增强,开发者需要将AI集成到实际业务工作流中,而不仅仅是简单的对话。Claude Code SDK提供了一种编程式、无头的方式,使开发者能够构建自动化、可扩展的AI代理系统。本教程针对的痛点包括:如何管理代理的长期记忆、如何执行外部脚本、如何适应不同受众、如何安全地规划复杂任务、如何复用提示词、如何添加审计和护栏,以及如何实现多代理协作。这些功能解决了从原型到生产部署的关键问题,适用于需要自动化决策支持、数据分析、报告生成和合规审计的企业场景。主要限制包括:SDK的无头模式与交互式CLI的差异(如计划模式的退出机制)、文件系统设置加载的显式配置要求,以及代理对数据源的自然偏好可能影响上下文使用。
核心原理
Claude Code SDK的核心原理是通过编程接口封装Claude Code的全部能力,实现功能对等。代理通过ClaudeSDKClient与模型交互,支持流式响应和工具调用。关键机制包括:
- CLAUDE.md记忆:作为持久上下文,在初始化时自动加载,提供项目级指令和背景,但代理会优先选择更细粒度的数据源,因此需要通过提示工程引导。
- Bash工具:允许代理执行Python脚本,实现复杂计算和数据分析,扩展了代理的原生能力。
- 输出风格:通过修改系统提示,调整输出风格以适应不同受众,存储在
.claude/output-styles/目录。 - 计划模式:通过
permission_mode="plan"启用,代理只制定计划不执行,计划可通过多种方式捕获和持久化。 - 斜杠命令:预定义的提示词模板,通过
/command触发,展开为完整提示词,支持参数。 - 钩子:在工具调用前后执行确定性脚本,用于审计和护栏,配置在
settings.local.json中。 - 子代理:通过Task工具委派任务给专业子代理,每个子代理有独立上下文和指令,实现并行化和专业化。
数据流:用户查询→主代理(结合CLAUDE.md上下文)→调用工具(Bash、Task等)→子代理执行→结果返回主代理→输出(按输出风格格式化)。组件关系:主代理协调子代理,钩子监控工具调用,斜杠命令简化用户输入,计划模式提供决策支持。
架构分析
系统架构设计遵循模块化和可扩展原则,核心组件包括:
- 代理核心:
ClaudeSDKClient封装了与模型的通信,支持异步流式处理。 - 配置层:通过
ClaudeAgentOptions配置模型、工作目录、允许工具、设置来源等。 - 文件系统设置:
CLAUDE.md、.claude/commands/、.claude/agents/、.claude/output-styles/、.claude/hooks/和settings.local.json共同构成代理的“大脑”和“肌肉”。 - 工具层:Bash、Read、Write、Edit、Task等工具提供外部交互能力。
- 子代理系统:通过
.claude/agents/定义,每个子代理有独立系统提示和工具,主代理通过Task工具动态委派。 - 钩子系统:在工具调用前后触发,实现审计、合规和护栏。
- 计划持久化:多源捕获机制(消息流、Write工具、Claude内部目录)确保计划可靠保存。
设计决策包括:使用setting_sources显式加载文件系统设置(避免隔离模式默认不加载);通过提示工程引导代理使用高层上下文;采用优先级回退的提取策略处理计划输出;将代理循环封装为send_query()函数,简化调用。这种架构支持从简单到复杂的渐进式构建,便于维护和扩展。
完整实战教程
from dotenv import load_dotenv
from utils.agent_visualizer import (
display_agent_response,
print_activity,
reset_activity_context,
visualize_conversation,
)
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
load_dotenv()
# Define the model to use throughout this notebook
# Using Opus 4.6 for its superior planning and reasoning capabilities
MODEL = "claude-opus-4-6"
print(f"📋 Notebook configured to use: {MODEL}")
引言
在笔记本00中,我们构建了一个简单的研究代理。在本笔记本中,我们将逐步介绍Claude Code SDK的关键功能,用于构建全面的代理。对于每个引入的功能,我们将解释:
- 是什么:该功能是什么
- 为什么:该功能能做什么,以及为什么你会想使用它
- 如何:展示如何使用它的最小实现
如果你熟悉Claude Code,你会注意到SDK带来了功能对等,并使你能够以编程的无头方式利用Claude Code的所有能力。
场景
在整个笔记本中,我们将为一家拥有50名员工、刚刚筹集了1000万美元A轮融资的初创公司构建一个AI参谋长。CEO需要数据驱动的洞察,以在积极增长与财务可持续性之间取得平衡。
我们最终的参谋长代理将:
- 协调不同领域的专业子代理
- 聚合来自多个来源的洞察
- 提供带有可操作建议的高管摘要
基本特性
功能 0:通过 CLAUDE.md 实现记忆
是什么:CLAUDE.md 文件充当代理的持久记忆和指令。当项目目录中存在该文件时,Claude Code 会在初始化代理时自动读取并整合此上下文。
为什么:无需在每次交互中重复提供项目上下文、团队偏好或标准,你可以在 CLAUDE.md 中一次性定义它们。这确保了行为的一致性,并通过避免冗余解释来减少 token 使用量。
如何操作:
- 在工作目录中放置一个
CLAUDE.md文件——在我们的示例中为:chief_of_staff_agent/CLAUDE.md - 将 ClaudeSDKClient 的
cwd参数设置为指向你的CLAUDE.md文件所在的目录 - 当你希望代理优先使用高层上下文而非详细数据文件时,使用明确的提示来引导代理
重要行为说明:当 CLAUDE.md 和详细数据文件(如 CSV)同时可用时,代理可能更倾向于读取更细粒度的数据源以提供精确答案。这是预期行为——代理自然会寻求权威数据。为确保代理使用高层的 CLAUDE.md 上下文,请使用明确的提示指令(参见下面的示例)。这教会我们一个重要经验:CLAUDE.md 提供的是*上下文和指导*,而非对数据源的硬性约束。
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
cwd="chief_of_staff_agent", # Points to subdirectory with our CLAUDE.md
setting_sources=["project"],
)
) as agent:
await agent.query("What's our current runway?")
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)
# Display the response with HTML rendering
display_agent_response(messages)
# With this prompt, the agent should use CLAUDE.md values: ~$500K burn, 20 months runway
理解 Agent 的数据源偏好
刚才发生了什么:
通过在提示词中添加 ,我们引导 Agent 依赖 CLAUDE.md 上下文,而不是从 CSV 文件中寻找更细粒度的数据。
关键见解:
- CLAUDE.md 作为上下文,而非约束:当你设置
cwd时,CLAUDE.md 文件会作为背景上下文加载。然而,Agent 会自然地寻找可用的最权威数据源。如果存在详细的 CSV 文件,Agent 可能会为了精确性而优先选择它们。
- 提示工程很重要:提示词中的“来自上下文的高层财务数字”向 Agent 表明,你希望从 CLAUDE.md 中获取简化的执行摘要($500K 消耗,20 个月跑道),而不是来自 financial_data/burn_rate.csv 的精确逐月数据($525K 总消耗,$235K 净消耗)。
- 架构设计选择:这种行为在生产系统中实际上是可取的——你希望 Agent 找到最佳数据源。CLAUDE.md 应包含:
- 高层上下文和策略
- 公司信息和标准
- 指向详细数据存储位置的指针
- 关于何时使用高层数字与详细数字的指南
- 现实世界模式:将 CLAUDE.md 视为“入职文档”,为 Agent 提供方向,而详细文件则是 Agent 在需要精确性时可以查询的“源系统”。
功能 1:用于执行 Python 脚本的 Bash 工具
是什么:Bash 工具允许你的智能体(除其他功能外)直接运行 Python 脚本,从而能够访问程序性知识、执行复杂计算、进行数据分析以及实现其他超出智能体原生能力的集成。
为什么:我们的 Chief of Staff 可能需要处理数据文件、运行财务模型或基于这些数据生成可视化。这些都是使用 Bash 工具的绝佳场景。
怎么做:将你的 Python 脚本放在智能体可以访问的位置,并添加一些上下文说明它们是什么以及如何调用。如果脚本是为你的 Chief of Staff 智能体准备的,请将此上下文添加到其 CLAUDE.md 文件中;如果脚本是为某个子智能体准备的,请将所述上下文添加到其 MD 文件中(稍后会有更多细节)。在本教程中,我们在 chief_of_staff_agent/scripts 中添加了五个示例:
hiring_impact.py:计算新工程招聘如何影响烧钱率、跑道和现金状况。对于financial-analyst子智能体来说,这是模拟招聘场景与每月 50 万美元烧钱率和 20 个月跑道的关键。talent_scorer.py:使用加权标准对候选人的技术技能、经验、文化契合度和薪资期望进行评分。这是recruiter子智能体根据 TechStart 的高级工程师基准(18 万至 22 万美元)对工程候选人进行排名的核心工具。simple_calculation.py:执行快速财务计算,如跑道、烧钱率和季度指标。这是 Chief of Staff 无需复杂建模即可获得即时指标的实用脚本。financial_forecast.py:根据当前 240 万美元 ARR 以每月 15% 的速度增长,模拟 ARR 增长情景(基准/乐观/悲观)。对于financial-analyst预测 B 轮准备情况并验证 3000 万美元融资目标至关重要。decision_matrix.py:为战略选择(如 SmartDev 收购或办公室扩张)创建加权决策矩阵。帮助 Chief of Staff 系统性地评估涉及多个利益相关者和标准的复杂决策。
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
allowed_tools=["Bash", "Read"],
cwd="chief_of_staff_agent", # Points to subdirectory where our agent is defined
)
) as agent:
await agent.query(
"Use your simple calculation script with a total runway of 2904829 and a monthly burn of 121938."
)
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)
# Display the response with HTML rendering
display_agent_response(messages)
功能 2:输出风格
是什么:输出风格允许你针对不同的受众使用不同的输出风格。每种风格都在一个 Markdown 文件中定义。
为什么:你的智能体可能会被不同专业水平的人使用,或者他们可能有不同的优先级。你的输出风格可以帮助区分这些群体,而无需创建单独的智能体。
如何操作:
- 在
chief_of_staff_agent/.claude/output-styles/中为每种风格配置一个 Markdown 文件。例如,查看.claude/output-styles/executive.md中的高管输出风格。输出风格通过简单的 frontmatter 定义,包含两个字段:名称和描述。注意:确保 frontmatter 中的名称与文件名完全匹配(区分大小写)。
重要提示:输出风格会修改 Claude Code 底层的系统提示,省略掉专注于软件工程的部分,并让你在软件工程工作之外针对特定用例获得更多控制。
SDK 配置说明:与斜杠命令(在功能 4 中介绍)类似,输出风格存储在文件系统的.claude/output-styles/目录中。为了让 SDK 加载这些文件,你必须在ClaudeAgentOptions中包含setting_sources=["project"]。settings参数告诉 SDK 使用*哪种*风格,但setting_sources是实际*加载*风格定义所必需的。此要求是在调试后续部分时发现的,并且适用于所有基于文件系统的设置。
messages_executive = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
cwd="chief_of_staff_agent",
settings='{"outputStyle": "executive"}',
# IMPORTANT: setting_sources must include "project" to load output styles from .claude/output-styles/
# Without this, the SDK does NOT load filesystem settings (output styles, slash commands, etc.)
setting_sources=["project"],
)
) as agent:
await agent.query("Tell me in two sentences about your writing output style.")
async for msg in agent.receive_response():
print_activity(msg)
messages_executive.append(msg)
messages_technical = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
cwd="chief_of_staff_agent",
settings='{"outputStyle": "technical"}',
setting_sources=["project"],
)
) as agent:
await agent.query("Tell me in two sentences about your writing output style.")
async for msg in agent.receive_response():
print_activity(msg)
messages_technical.append(msg)
# Display executive style response
display_agent_response(messages_executive)
# Technical output style - detailed, implementation-focused
display_agent_response(messages_technical, title="Technical Style")
功能 3:计划模式——不执行操作的策略规划
是什么:计划模式指示代理创建详细的执行计划,而不执行任何操作。代理分析需求、提出解决方案并概述步骤,但不修改文件、执行命令或进行更改。
为什么:复杂的任务受益于预先规划,以减少错误、便于审查并改善协调。在规划阶段之后,代理将有一条贯穿执行过程的主线。
如何操作:只需设置 permission_mode="plan"
计划持久化:由于计划是用于审查和决策的宝贵工件,我们将演示如何捕获并将其保存为持久的 Markdown 文件。这使利益相关者能够在批准执行之前审查计划。
注意:此功能在 Claude Code 中表现出色,但仍需完全适配使用 SDK 的无头应用程序。也就是说,代理会尝试调用其ExitPlanMode()工具,该工具仅在交互模式下相关。在这种情况下,您可以发送后续查询,并设置continue_conversation=True,让代理在上下文中执行其计划。
# =============================================================================
# Plan Mode Helper Functions
# =============================================================================
# These utilities handle the various ways an agent might output its plan.
# Since agents can output plans via direct text, Write tool, or Claude's
# internal plan directory, we need robust extraction from multiple sources.
import glob as glob_module
import os
import re
from datetime import datetime
from pathlib import Path
from typing import Any
def extract_plan_from_xml(text: str | None, min_length: int = 200) -> str | None:
"""
Extract content between <plan> tags from text.
Args:
text: The text to search for plan content
min_length: Minimum character count for valid plan (prevents empty matches)
Returns:
Extracted plan content, or None if not found/too short
"""
if not text:
return None
match = re.search(r"<plan>(.*?)</plan>", text, re.DOTALL)
if match:
extracted = match.group(1).strip()
if len(extracted) > min_length:
return extracted
return None
def extract_plan_from_messages(
plan_content: list[str], min_fallback_length: int = 500
) -> tuple[str | None, str | None]:
"""
Try to extract plan from captured message stream content.
Args:
plan_content: List of text blocks captured during streaming
min_fallback_length: Minimum length for fallback (no XML tags)
Returns:
Tuple of (plan_text, source_description)
"""
combined_text = "\n\n".join(plan_content)
# First try: XML tags
plan = extract_plan_from_xml(combined_text)
if plan:
return plan, "message stream"
# Fallback: Use raw content if substantial
if len(combined_text.strip()) > min_fallback_length:
return combined_text.strip(), "full message content (fallback)"
return None, None
def extract_plan_from_write_tool(
write_contents: list[str], min_fallback_length: int = 500
) -> tuple[str | None, str | None]:
"""
Try to extract plan from captured Write tool calls.
Args:
write_contents: List of content strings from Write tool calls
min_fallback_length: Minimum length for fallback (no XML tags)
Returns:
Tuple of (plan_text, source_description)
"""
for content in write_contents:
# Try XML extraction first
plan = extract_plan_from_xml(content)
if plan:
return plan, "Write tool capture"
# Fallback: substantial content without tags
if content and len(content.strip()) > min_fallback_length:
return content.strip(), "Write tool capture (no XML tags)"
return None, None
def extract_plan_from_claude_dir(
max_age_seconds: int = 300, min_fallback_length: int = 500
) -> tuple[str | None, str | None]:
"""
Check Claude's internal plan directory for recently created plans.
Args:
max_age_seconds: Maximum age of plan file to consider (default: 5 minutes)
min_fallback_length: Minimum length for fallback (no XML tags)
Returns:
Tuple of (plan_text, source_description)
"""
claude_plans_dir = os.path.expanduser("~/.claude/plans")
if not os.path.exists(claude_plans_dir):
return None, None
# Find most recent plan file
plan_files = sorted(
glob_module.glob(os.path.join(claude_plans_dir, "*.md")),
key=os.path.getmtime,
reverse=True,
)
if not plan_files:
return None, None
most_recent = plan_files[0]
file_age = datetime.now().timestamp() - os.path.getmtime(most_recent)
if file_age > max_age_seconds:
return None, None
with open(most_recent) as f:
content = f.read()
filename = os.path.basename(most_recent)
# Try XML extraction first
plan = extract_plan_from_xml(content)
if plan:
return plan, f"Claude plan file ({filename})"
# Fallback: substantial content without tags
if len(content.strip()) > min_fallback_length:
return content.strip(), f"Claude plan file ({filename}, no XML tags)"
return None, None
def save_plan_to_file(
plan_content: str,
plan_source: str,
model_name: str,
prompt_summary: str,
output_dir: str = "chief_of_staff_agent/plans",
title: str = "Agent Plan: Engineering Restructure for AI Focus",
) -> Path:
"""
Save extracted plan to a timestamped markdown file.
Args:
plan_content: The plan text to save
plan_source: Description of where plan was extracted from
model_name: The model used to generate the plan
prompt_summary: Brief description of the original prompt
output_dir: Directory to save plan files
title: Title for the plan document
Returns:
Path to the saved plan file
"""
plans_dir = Path(output_dir)
plans_dir.mkdir(exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
plan_file = plans_dir / f"plan_{timestamp}.md"
with open(plan_file, "w") as f:
f.write(f"# {title}\n\n")
f.write(f"**Created:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"**Prompt:** {prompt_summary}\n")
f.write(f"**Model:** {model_name}\n")
f.write(f"**Plan Source:** {plan_source}\n\n")
f.write("---\n\n")
f.write(plan_content)
f.write("\n\n---\n\n")
f.write("*This plan was generated in plan mode and has not been executed.*\n")
return plan_file
def capture_message_content(
msg: Any,
plan_content: list[str],
write_tool_content: list[str],
write_tool_paths: list[str],
) -> None:
"""
Process a streaming message and capture relevant plan content.
This function extracts content from three potential sources:
1. Text blocks in message content
2. Write tool call parameters
3. Final result attribute
Args:
msg: The message object from the agent stream
plan_content: List to append text content to
write_tool_content: List to append Write tool content to
write_tool_paths: List to append Write tool file paths to
"""
# Source 1: Text blocks from message content
if hasattr(msg, "content"):
for block in msg.content:
if hasattr(block, "text"):
plan_content.append(block.text)
# Source 2: Write tool calls
if hasattr(block, "type") and block.type == "tool_use":
if hasattr(block, "name") and block.name == "Write":
if hasattr(block, "input") and isinstance(block.input, dict):
if "content" in block.input:
write_tool_content.append(block.input["content"])
if "file_path" in block.input:
write_tool_paths.append(block.input["file_path"])
# Source 3: Final result
if hasattr(msg, "result") and msg.result:
plan_content.append(msg.result)
print("✅ Plan Mode helper functions loaded")
# =============================================================================
# Plan Mode Configuration
# =============================================================================
# Note: MODEL is defined in cell-0 as "claude-opus-4-6"
# Opus excels at complex planning tasks
# The prompt is carefully crafted to:
# 1. Provide explicit context (since Opus prefers explicit information)
# 2. Request XML-tagged output for reliable extraction
# 3. Prevent file-writing so we can capture the plan programmatically
PLAN_PROMPT = """Restructure our engineering team for AI focus.
**CONTEXT (from CLAUDE.md):**
You are the Chief of Staff for TechStart Inc, a 50-person B2B SaaS startup that raised $10M Series A.
- Current engineering team: 25 people (Backend: 12, Frontend: 8, DevOps: 5)
- Monthly burn rate: ~$500K, Runway: 20 months
- Senior Engineer compensation: $180K-$220K + equity
**CRITICAL OUTPUT INSTRUCTIONS:**
1. **DO NOT use the Write tool** - Output your plan directly in your response text
2. **DO NOT save to any files** - I will handle saving the plan myself
3. **Wrap your ENTIRE plan inside `<plan> </plan>` XML tags** in your response
**Required Format:**
<plan>
[Your complete restructuring plan here - include all sections, timelines, budgets, and recommendations]
</plan>
**IMPORTANT:**
- The plan content MUST appear directly in your response between the XML tags
- Do NOT use Write, Edit, or any file-saving tools
- You may research and analyze before outputting, but the final plan must be in your response text
- Include: team structure, hiring recommendations, timeline, budget impact, and success metrics
- Use the company context provided above - do NOT ask clarifying questions"""
print(f"📋 Plan Mode configured with model: {MODEL}")
print(f"📝 Prompt length: {len(PLAN_PROMPT):,} characters")
# =============================================================================
# Execute Plan Mode Agent
# =============================================================================
# Run the agent with plan mode enabled. The agent will create a detailed plan
# but won't execute any actions. We capture content from multiple sources
# to handle different agent behaviors.
# Initialize capture lists
messages = []
plan_content = [] # Text from message stream
write_tool_content = [] # Content from Write tool calls
write_tool_paths = [] # Paths from Write tool calls
# Run the agent in plan mode
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
permission_mode="plan",
cwd="chief_of_staff_agent",
)
) as agent:
await agent.query(PLAN_PROMPT)
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)
# Capture content from this message
capture_message_content(msg, plan_content, write_tool_content, write_tool_paths)
print(f"\n✅ Agent completed. Captured {len(plan_content)} content blocks.")
# =============================================================================
# Extract and Save the Plan
# =============================================================================
# Try multiple sources in priority order to find the plan content.
# This handles different agent behaviors robustly.
final_plan = None
plan_source = None
# Priority 1: Message stream (preferred - direct from agent response)
final_plan, plan_source = extract_plan_from_messages(plan_content)
# Priority 2: Write tool captures (if agent saved despite instructions)
if not final_plan and write_tool_content:
final_plan, plan_source = extract_plan_from_write_tool(write_tool_content)
# Priority 3: Claude's internal plan directory (safety net)
if not final_plan:
final_plan, plan_source = extract_plan_from_claude_dir()
# Report results
if final_plan:
print(f"✅ Plan extracted from: {plan_source}")
print(f" Plan length: {len(final_plan):,} characters")
# Save to file
plan_file = save_plan_to_file(
plan_content=final_plan,
plan_source=plan_source,
model_name=MODEL,
prompt_summary="Restructure our engineering team for AI focus.",
)
print(f"\n📁 Plan saved to: {plan_file}")
else:
error_msg = "Could not extract plan content from any source!\n"
error_msg += " Sources checked: message stream, Write tool, ~/.claude/plans/"
if write_tool_paths:
error_msg += f"\n Write tool attempted to save to: {write_tool_paths}"
print(f"❌ ERROR: {error_msg}")
raise RuntimeError(f"Plan extraction failed: {error_msg}")
# Display the plan result with styled HTML
display_agent_response(messages, title="Engineering Restructure Plan")
执行已保存的计划
如上所述,代理将在创建计划后停止。保存的计划文件作为利益相关者的审查工件。
审查后执行计划:
- 在
chief_of_staff_agent/plans/plan_*.md中查看已保存的计划 - 如果批准,发送带有
continue_conversation=True的新查询,并移除permission_mode="plan"以执行
此工作流实现了“计划 → 审查 → 批准 → 执行”的循环,非常适合组织重组或重大基础设施变更等高风险的决策。
计划持久化如何工作
在上面的代码中,我们实现了一个健壮的多源计划捕获机制,用于处理 Plan 模式代理可能输出计划的各种方式:
挑战:
当使用 permission_mode="plan" 时,代理可能以不同方式输出计划:
- 消息流中的直接文本输出(理想情况)
- Write 工具保存到
~/.claude/plans/(Claude 的内部计划系统) - Write 工具保存到自定义路径
我们的捕获机制通过基于优先级的回退系统处理所有三种场景:
来源优先级(按顺序):
- 消息流(首选)
- 在流式传输期间从
msg.content捕获文本块 - 提取
<plan></plan>XML 标签之间的内容 - 这是最干净的方法,因为内容直接来自响应
- Write 工具捕获
- 监视消息流中的 Write 工具调用
- 提取正在写入的
content参数 - 当代理尽管有提示指令仍决定保存时很有用
- Claude 的内部计划目录
- 检查
~/.claude/plans/中最近创建的计划文件(5 分钟内) - 读取并提取最新文件的内容
- 当其他方法失败时作为安全网
- 完整内容回退
- 如果未找到 XML 标签但存在大量内容(>500 字符),则直接使用
- 防止空计划文件,同时保留部分信息
关键实现细节:
def extract_plan_from_text(text):
"""Extract content between <plan> tags, return None if not found or empty."""
match = re.search(r'<plan>(.*?)</plan>', text, re.DOTALL)
if match:
extracted = match.group(1).strip()
# Validate minimum content length (a real plan should be substantial)
if len(extracted) > 200:
return extracted
return None
为什么内容验证很重要:
- 以前的版本可能会在提取“成功”但没有内容时生成空的计划文件
- 我们现在要求 XML 标记的内容至少 200 个字符
- 这可以防止正则表达式匹配空或琐碎内容时的误报
直接输出的提示工程:
提示明确指示代理:
- 不要使用 Write 工具 - 防止文件系统绕行
- 直接在响应中输出 - 确保内容通过消息流传递
- 使用 XML 标签 - 能够从可能冗长的响应中干净地提取
这种方法为你提供:
- 可靠性:无论代理行为如何,计划都会被捕获
- 透明度:保存的文件指示使用了哪个来源
- 审计跟踪:所有计划的历史记录,带有时间戳和来源元数据
- 调试:提取失败时清晰的错误消息
让我们查看保存的计划:
# Display the saved plan with markdown rendering
from IPython.display import Markdown, display
# Show the plan with proper markdown formatting
display(Markdown(f"## 📋 Saved Plan Preview\n\n{final_plan}"))
print(f"\n📁 Full plan with metadata saved to: {plan_file}")
高级特性
功能 4:自定义斜杠命令
注意:斜杠命令对用户来说是语法糖,并非新的智能体能力
是什么:自定义斜杠命令是预定义的提示词模板,用户可以通过简写语法(例如 /budget-impact)触发。这些是面向用户的快捷方式,而非智能体能力。可以把它们想象成键盘快捷键,展开后变成完整、精心编写的提示词。
为什么:你的 Chief of Staff 将处理重复出现的高管问题。用户无需反复输入复杂的提示词,而是可以使用已经过验证的提示词。这提高了一致性和标准化程度。
怎么做:
- 在
.claude/commands/中定义一个 Markdown 文件。例如,我们在.claude/commands/slash-command-test.md中定义了一个。注意命令是如何定义的:frontmatter 包含两个字段(name 和 description),以及展开后的提示词,并带有包含查询中传入参数的可选功能。 - 你可以使用
$ARGUMENTS(用于完整参数字符串)或$1、$2等(用于位置参数)向提示词中添加参数。 - 用户在提示词中使用斜杠命令。
关键 SDK 配置:使用 SDK 时,必须在ClaudeAgentOptions中设置setting_sources=["project"],斜杠命令才能生效。默认情况下,SDK 以隔离模式运行,不会加载文件系统设置(斜杠命令、CLAUDE.md、子代理、钩子等)。这与交互式使用 Claude Code 时自动加载这些设置不同。
# User types: "/slash-command-test this is a test"
# -> behind the scenes EXPANDS to the prompt in .claude/commands/slash-command-test.md
# In this case the expanded prompt says to simply reverse the sentence word wise
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
cwd="chief_of_staff_agent",
# IMPORTANT: setting_sources must include "project" to load slash commands from .claude/commands/
# Without this, the SDK does NOT load filesystem settings (slash commands, CLAUDE.md, etc.)
setting_sources=["project"],
)
) as agent:
await agent.query("/slash-command-test this is a test")
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)
display_agent_response(messages, title="Slash Command Result")
功能 5:钩子 - 自动化确定性操作
是什么:钩子是 Python 脚本,你可以设置它们在特定工具调用之前(pre)或之后(post)等事件中自动执行。钩子以确定性方式运行,非常适合用于验证和审计跟踪。
为什么:想象一下这样的场景:你想确保你的代理有一些护栏(例如,防止危险操作),或者你想要一个审计跟踪。钩子与代理结合使用非常理想,既能给代理足够的自由来完成任务,同时又能确保代理以安全的方式行事。
如何操作:
- 在
.claude/hooks/中定义钩子脚本 -> _什么_ 行为应该在钩子触发时执行 - 在
.claude/settings.local.json中定义钩子配置 -> _何时_ 应该触发钩子 - 在这种情况下,我们的钩子被配置为监视特定的工具调用(Bash、Write、Edit)
- 当这些工具被调用时,钩子脚本会在工具完成后运行(PostToolUse)
SDK 配置说明:在.claude/settings.local.json中配置的钩子需要setting_sources=["project", "local"]。SDK 区分三种设置来源:
-"project"→.claude/settings.json(版本控制,团队共享)
-"local"→.claude/settings.local.json(gitignored,本地设置,如钩子)
-"user"→~/.claude/settings.json(全局用户设置)
由于我们的钩子在settings.local.json中,我们必须将"local"包含在setting_sources中。
示例:用于合规性的报告跟踪
一个钩子,用于记录对财务报告的写入/编辑操作,以供审计和合规目的。
该钩子定义在 chief_of_staff_agent/.claude/hooks/report-tracker.py 中,强制执行它的逻辑在 chief_of_staff_agent/.claude/settings.local.json 中:
"hooks": {
"PostToolUse": [
{
"matcher": "Write",
"hooks": [
{
"type": "command",
"command": "$CLAUDE_PROJECT_DIR/.claude/hooks/report-tracker.py"
}
]
},
{
"matcher": "Edit",
"hooks": [
{
"type": "command",
"command": "$CLAUDE_PROJECT_DIR/.claude/hooks/report-tracker.py"
}
]
}
]
}
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
cwd="chief_of_staff_agent",
allowed_tools=["Bash", "Write", "Edit", "MultiEdit"],
# IMPORTANT: setting_sources must include BOTH "project" AND "local" to load hooks
# - "project" loads .claude/settings.json (shared settings, CLAUDE.md, slash commands)
# - "local" loads .claude/settings.local.json (where hooks are configured)
setting_sources=["project", "local"],
)
) as agent:
await agent.query(
"Create a quick Q2 financial forecast report with our current burn rate and runway projections. Save it to our /output_reports folder."
)
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)
# The hook will track this in audit/report_history.json
display_agent_response(messages, title="Q2 Financial Forecast")
如果你现在导航到 ./chief_of_staff_agent/audit/report_history.json,你会发现它已经记录了该代理创建和/或修改报告的操作。生成的报告本身可以在 ./chief_of_staff_agent/output_reports/ 中找到。
功能 6:通过任务工具实现子代理
是什么:任务工具使您的代理能够将专业工作委派给其他子代理。这些子代理各自拥有自己的指令、工具和专业知识。
为什么:添加子代理开启了许多可能性:
- 专业化:每个子代理都是其领域的专家
- 独立上下文:子代理拥有自己的对话历史和工具
- 并行化:多个子代理可以同时处理不同方面的工作。
如何操作:
- 将
"Task"添加到 allowed_tools 中 - 使用系统提示指导您的代理如何委派任务(您也可以在其 CLAUDE.md 中更一般地定义这一点)
- 在
.claude/agents/中为每个代理创建一个 markdown 文件。例如,查看.claude/agents/financial-analyst.md中的文件,并注意如何通过这样一个简单直观的 markdown 文件来定义(子)代理:包含三个字段(名称、描述和工具)的 frontmatter 及其系统提示。描述对于主要的总参谋长代理了解何时调用每个子代理非常有用。
可视化增强:我们的 print_activity() 和 visualize_conversation() 工具已得到增强,可以清晰显示子代理操作:
- 🚀 表示正在委派给子代理(带有子代理名称)
- 📎 表示子代理正在使用的工具(缩进以显示视觉层次)
- 视觉分隔符清晰标记子代理委派和完成的边界
- 任务描述和提示显示在对话时间线中
# Reset the subagent tracking context before starting a new query
# This ensures clean state for activity display
reset_activity_context()
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model=MODEL,
allowed_tools=["Task"], # this enables our Chief agent to invoke subagents
system_prompt="Delegate financial questions to the financial-analyst subagent. Do not try to answer these questions yourself.",
cwd="chief_of_staff_agent",
setting_sources=["project", "local"],
)
) as agent:
await agent.query("Should we hire 5 engineers? Analyze the financial impact.")
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)
display_agent_response(messages, title="Hiring Impact Analysis")
visualize_conversation(messages)
在这里,当我们的主代理决定使用子代理时,它将:
- 调用 Task 工具,并传入如下参数:
- Task 工具在单独的上下文中执行子代理
- 将结果返回给主 Chief of Staff 代理以继续处理
{
"description": "分析招聘影响",
"prompt": "分析招聘5名工程师的财务影响...",
"subagent_type": "financial-analyst"
}
综合运用
现在,让我们把之前看到的所有功能整合在一起。我们将让我们的智能体评估招聘3名高级工程师的财务影响,并将其见解写入 output_reports/hiring_decision.md。这将展示上述所有功能:
- Bash 工具:用于执行
hiring_impact.py脚本,以确定招聘新工程师的影响 - 记忆:读取目录中的
CLAUDE.md作为上下文,以了解当前的预算、资金储备、收入和其他相关信息 - 输出风格:在
chief_of_staff_agent/.claude/output-styles中定义的不同输出风格 - 自定义斜杠命令:使用快捷键
/budget-impact,该快捷键会扩展为chief_of_staff_agent/.claude/commands中定义的完整提示词 - 子代理:我们的
/budget_impact命令引导参谋长代理调用在chief_of_staff_agent/.claude/agents中定义的财务分析师子代理 - 钩子:钩子在
chief_of_staff_agent/.claude/hooks中定义,并在chief_of_staff_agent/.claude/settings.local.json中配置 - 如果我们的某个代理正在更新财务报告,钩子应将此编辑/写入活动记录在
chief_of_staff_agent/audit/report_history.json日志文件中 - 如果财务分析师子代理将调用
hiring_impact.py脚本,这将被记录在chief_of_staff_agent/audit/tool_usage_log.json日志文件中
- 计划模式:如果你希望参谋长在采取任何行动之前先制定一个计划供你批准,请取消下面注释行的注释
为了准备好这一切,我们将代理循环封装在一个 Python 文件中,类似于我们在上一个笔记本中所做的。请查看 chief_of_staff_agent 子目录中的 agent.py 文件。
总而言之,我们的 send_query() 函数接受4个参数(提示词、继续对话、权限模式和输出风格),其他所有内容都在代理文件中设置,即:系统提示词、最大轮数、允许的工具和工作目录。
为了更好地可视化这一切是如何组合在一起的,请查看这些 Claude 为我们制作的流程和架构图 :)
from chief_of_staff_agent.agent import send_query
reset_activity_context()
result, messages = await send_query(
"/budget-impact hiring 3 senior engineers. Save your insights by updating the 'hiring_decision.md' file in /output_reports or creating a new file there",
# permission_mode="plan", # Enable this to use planning mode
output_style="executive",
)
visualize_conversation(messages)
结论
我们已经演示了 Claude Code SDK 如何使您能够构建具有企业级功能的复杂多智能体系统。从使用 Bash 工具的基本脚本执行开始,我们逐步引入了高级功能,包括使用 CLAUDE.md 的持久记忆、针对不同受众的自定义输出风格、战略规划模式、方便用户的斜杠命令、用于护栏的合规性钩子,以及用于专门任务的子智能体协调。
通过结合这些功能,我们创建了一个能够处理复杂高管决策工作流的 AI 幕僚长。该系统将财务分析委派给专门的子智能体,通过钩子维护审计跟踪,适应不同利益相关者的沟通风格,并提供由数据驱动分析支持的可操作见解。
这一高级智能体模式和多智能体编排的基础为您构建生产级企业系统做好了准备。在下一个笔记本中,我们将探讨如何通过模型上下文协议(MCP)服务器将我们的智能体连接到外部服务,从而大幅扩展其超越内置工具的能力。
下一步:02_The_observability_agent.ipynb - 了解如何通过 MCP 使用自定义集成和外部数据源扩展您的智能体。
常见问题
常见错误及排查方法:
- 斜杠命令或输出风格不生效:确保
setting_sources包含"project",否则SDK不会加载文件系统设置。 - 钩子不触发:钩子配置在
settings.local.json,需在setting_sources中同时包含"project"和"local"。 - 计划模式代理不退出:代理会调用
ExitPlanMode()工具,在无头模式下需发送后续查询并设置continue_conversation=True来执行计划。 - 代理忽略CLAUDE.md而读取CSV:这是预期行为,代理偏好精确数据。可通过明确提示(如“使用来自上下文的高层财务数字”)引导。
- 计划提取失败:检查是否使用了XML标签,确保内容长度超过200字符;若代理使用Write工具,需捕获Write调用;检查
~/.claude/plans/目录。 - 子代理未调用:确保
allowed_tools包含"Task",并在系统提示中明确委派策略。
生产环境优化:
- 使用
setting_sources明确加载所需设置,避免不必要的文件读取。 - 为钩子脚本添加错误处理和日志,避免影响主流程。
- 对计划提取逻辑进行单元测试,确保健壮性。
- 使用异步和流式处理提高响应速度。
- 监控token使用,优化提示词长度。
- 将代理配置版本化,便于回滚和审计。
企业应用扩展
本教程展示了Claude Code SDK的核心功能,但生产级系统还需考虑以下扩展:
- 模型上下文协议(MCP):通过MCP服务器连接外部服务(如数据库、API),大幅扩展代理能力。
- 可观测性:集成日志、追踪和监控,如使用LangSmith或自定义日志系统,记录代理决策和工具调用。
- 多代理编排:更复杂的编排模式,如层级代理、动态任务分配、结果聚合。
- 安全与合规:实现更细粒度的权限控制、敏感数据脱敏、审计日志的加密存储。
- 性能优化:缓存常用查询结果、并行执行独立子任务、使用更高效的模型。
- 用户界面:构建前端界面,使非技术用户能通过自然语言与代理交互。
- 持续学习:根据反馈调整CLAUDE.md和提示词,优化代理行为。
- 错误恢复:实现重试机制、降级策略和人工介入流程。
- 测试:编写集成测试和模拟测试,确保代理在边界情况下的稳定性。
总结
本教程面向有编程基础的开发者,深入讲解如何使用Claude Code SDK构建企业级AI代理。通过一个为初创公司CEO服务的AI参谋长案例,系统介绍了CLAUDE.md记忆、Bash工具、输出风格、计划模式、斜杠命令、钩子和子代理等核心功能。教程不仅展示代码,还剖析了每个功能的设计原理、适用场景和架构决策,并提供了完整的实践步骤、常见错误排查和生产优化建议,帮助开发者掌握构建复杂多代理系统的工程实践。
- --
内容来源与验证
- 内容等级:PREMIUM_A
- 质量等级:S_CANDIDATE
- 难度:高级
- 验证等级:V1
- 原文作者:rodrigo-olivares, JiriDeJonghe
- 原文日期:2025-09-12
- 来源许可:MIT
- 来源版本:
9dde2426171451e8dd7d6939fe398f97a1fb92e7 - 查看原文
站内还有更多完整实战教程
加入会员,解锁全部教程、完整代码与持续更新内容。
加入会员