构建代理 - 模式和最佳实践

设计模式、示例和构建健壮目标驱动代理的最佳实践。

先决条件： 使用 hive-create 完成代理结构。

实际示例：混合工作流

如何使用直接文件写入和可选MCP验证构建节点：

# 1. 首先写入文件（主要 - 使其可见）
node_code = '''
search_node = NodeSpec(
    id="search-web",
    node_type="event_loop",
    input_keys=["query"],
    output_keys=["search_results"],
    system_prompt="Search the web for: {query}. Use web_search, then call set_output to store results.",
    tools=["web_search"],
)
'''

Edit(
    file_path="exports/research_agent/nodes/__init__.py",
    old_string="# Nodes will be added here",
    new_string=node_code
)

# 2. 可选地使用MCP进行验证（次要 - 记账）
validation = mcp__agent-builder__test_node(
    node_id="search-web",
    test_input='{"query": "python tutorials"}',
    mock_llm_response='{"search_results": [...mock results...]}'
)

用户体验：

立即在编辑器中看到节点（来自步骤1）
获得验证反馈（来自步骤2）
如有需要，可以直接编辑文件

多轮交互模式

对于需要与用户进行多轮对话的代理，使用 client_facing=True 在event_loop节点上。

面向客户端的节点

面向客户端的节点将LLM输出流式传输到用户，并在对话轮流之间阻塞用户输入。这取代了旧的暂停/恢复模式。

# 面向客户端的节点，带有STEP 1/STEP 2提示模式
intake_node = NodeSpec(
    id="intake",
    name="Intake",
    description="Gather requirements from the user",
    node_type="event_loop",
    client_facing=True,
    input_keys=["topic"],
    output_keys=["research_brief"],
    system_prompt="""\
You are an intake specialist.

**STEP 1 — Read and respond (text only, NO tool calls):**
1. Read the topic provided
2. If it's vague, ask 1-2 clarifying questions
3. If it's clear, confirm your understanding

**STEP 2 — After the user confirms, call set_output:**
- set_output("research_brief", "Clear description of what to research")
"""),

# 内部节点无需用户交互即可运行
research_node = NodeSpec(
    id="research",
    name="Research",
    description="Search and analyze sources",
    node_type="event_loop",
    input_keys=["research_brief"],
    output_keys=["findings", "sources"],
    system_prompt="Research the topic using web_search and web_scrape...",
    tools=["web_search", "web_scrape", "load_data", "save_data"],
)

工作原理：

面向客户端的节点将LLM文本流式传输到用户，并在每个响应后阻塞输入
用户输入通过 node.inject_event(text) 注入
当LLM调用 set_output 生成结构化输出时，裁判评估并接受
内部节点（非面向客户端）运行整个循环而不会阻塞
set_output 是一个合成工具 —— 一个只有 set_output 调用的回合（没有真正的工具）会触发用户输入阻塞

STEP 1/STEP 2模式： 总是用明确的阶段构建面向客户端的提示。STEP 1是仅限文本的对话。STEP 2在用户确认后调用 set_output。这可以防止LLM在用户响应之前过早地调用 set_output。

使用client_facing的情况

场景	client_facing	为什么
收集用户需求	Yes	需要用户输入
人工审核/批准检查点	Yes	需要人类决策
数据处理（扫描、评分）	No	自主运行
报告生成	No	不需要用户输入
行动前的最终确认	Yes	需要明确的批准

遗留说明： pause_nodes / entry_points 模式仍然适用于向后兼容，但推荐对新代理使用 client_facing=True。

基于边缘的路由和反馈循环

条件边缘路由

从同一来源的多个条件边缘替换了旧的 router 节点类型。每个边缘检查节点输出上的条件。

# 具有互斥输出的节点
review_node = NodeSpec(
    id="review",
    name="Review",
    node_type="event_loop",
    client_facing=True,
    output_keys=["approved_contacts", "redo_extraction"],
    nullable_output_keys=["approved_contacts", "redo_extraction"],
    max_node_visits=3,
    system_prompt="Present the contact list to the operator. If they approve, call set_output('approved_contacts', ...). If they want changes, call set_output('redo_extraction', 'true').",
)

# 正向边缘（优先级高，首先评估）
EdgeSpec(
    id="review-to-campaign",
    source="review",
    target="campaign-builder",
    condition=EdgeCondition.CONDITIONAL,
    condition_expr="output.get('approved_contacts') is not None",
    priority=1,
)

# 反馈边缘（优先级低，评估后正向边缘）
EdgeSpec(
    id="review-feedback",
    source="review",
    target="extractor",
    condition=EdgeCondition.CONDITIONAL,
    condition_expr="output.get('redo_extraction') is not None",
    priority=-1,
)

关键概念：

nullable_output_keys: 列出可能未设置的输出键。节点每次执行设置其中之一互斥键。
max_node_visits: 反馈目标（提取器）上必须大于1才能重新执行。默认是1。
priority: 正数 = 正向边缘（首先评估）。负数 = 反馈边缘。执行器首先尝试正向边缘；如果没有匹配，回退到反馈边缘。

路由决策表

模式	旧方法	新方法
条件分支	`router` 节点	带有 `condition_expr` 的条件边缘
二元批准/拒绝	`pause_nodes` + 恢复	`client_facing=True` + `nullable_output_keys`
拒绝时回环	手动 entry_points	带有 `priority=-1` 的反馈边缘
多路路由	带有 routes 字典的路由器	带有优先级的多个条件边缘

裁判模式

核心原则：裁判是接受决策的唯一机制。 永远不要添加框架门控来补偿LLM行为。如果LLM过早地调用 set_output，请修复系统提示或使用自定义裁判。避免的反模式：

输出回滚逻辑
_user_has_responded 标志
提前拒绝设置_output
将交互协议注入系统提示

裁判控制event_loop节点的循环退出。根据验证需求选择。

隐式裁判（默认）

当没有配置裁判时，隐式裁判在以下情况下接受：

LLM完成响应且没有工具调用
所有必需的输出键都已通过 set_output 设置

最适合简单的节点，其中“所有输出键都设置”就足够了验证。

SchemaJudge

根据Pydantic模型验证输出。当您需要结构验证时使用。

from pydantic import BaseModel

class ScannerOutput(BaseModel):
    github_users: list[dict]  # 必须是用户对象列表

class SchemaJudge:
    def __init__(self, output_model: type[BaseModel]):
        self._model = output_model

    async def evaluate(self, context: dict) -> JudgeVerdict:
        missing = context.get("missing_keys", [])
        if missing:
            return JudgeVerdict(
                action="RETRY",
                feedback=f"Missing output keys: {missing}. Use set_output to provide them.",
            )
        try:
            self._model.model_validate(context["output_accumulator"])
            return JudgeVerdict(action="ACCEPT")
        except ValidationError as e:
            return JudgeVerdict(action="RETRY", feedback=str(e))

使用哪种裁判

裁判	使用时	示例
隐式（无）	输出键足够验证	简单的数据提取
SchemaJudge	需要结构验证的输出	API响应解析
自定义	特定于域的验证逻辑	分数必须是0.0-1.0

扇出/扇入（并行执行）

从同一来源的多个ON_SUCCESS边缘触发并行执行。所有分支通过 asyncio.gather() 并发运行。

# Scanner并行扇出到Profiler和Scorer
EdgeSpec(id="scanner-to-profiler", source="scanner", target="profiler",
         condition=EdgeCondition.ON_SUCCESS)
EdgeSpec(id="scanner-to-scorer", source="scanner", target="scorer",
         condition=EdgeCondition.ON_SUCCESS)

# 两者都扇入到Extractor
EdgeSpec(id="profiler-to-extractor", source="profiler", target="extractor",
         condition=EdgeCondition.ON_SUCCESS)
EdgeSpec(id="scorer-to-extractor", source="scorer", target="extractor",
         condition=EdgeCondition.ON_SUCCESS)

要求：

并行event_loop节点必须具有不相交的输出键（没有键由两者写入）
只有一个并行分支可以包含 client_facing 节点
扇入节点从所有完成的分支接收输出，在共享内存中

上下文管理模式

分层压缩

EventLoopNode自动使用分层压缩管理上下文窗口使用情况：

修剪 — 旧工具结果被紧凑的占位符替换（零成本，无需LLM调用）
正常压缩 — LLM总结旧消息
积极压缩 — 只保留最近的邮件+摘要
紧急 — 硬重置，保留工具历史

溢出模式

框架自动截断大型工具结果，并将完整内容保存到溢出目录。LLM接收截断消息，并带有使用 load_data 读取完整结果的说明。

对于显式数据管理，请使用数据工具（真正的MCP工具，不是合成的）：

# save_data, load_data, list_data_files, serve_file_to_user是真正的MCP工具
# data_dir由框架自动注入 — LLM从未看到它

# 保存大型结果
save_data(filename="sources.json", data=large_json_string)

# 带分页的读取（基于行的偏移/限制）
load_data(filename="sources.json", offset=0, limit=50)

# 列出可用文件
list_data_files()

# 将文件作为可点击链接提供给用户
serve_file_to_user(filename="report.html", label="Research Report")

将数据工具添加到处理大型工具结果的节点：

research_node = NodeSpec(
    ...
    tools=["web_search", "web_scrape", "load_data", "save_data", "list_data_files"],
)

data_dir是框架上下文参数 — 在调用时自动注入。GraphExecutor.execute() 每次执行通过 ToolRegistry.set_execution_context(data_dir=...) 设置它（使用 contextvars 保证并发安全），确保它与会话范围的溢出目录匹配。

反模式

不要做什么

不要依赖 export_graph — 立即写文件，而不是在结束时
不要隐藏会话中的代码 — 组件一旦批准就写入文件
不要等待写文件 — 从第一步开始就可以看到代理
不要批量处理一切 — 逐步写入，一次一个组件
不要创建太多薄节点 — 偏好较少，更丰富的节点（见下文）
不要为LLM行为添加框架门控 — 修复提示或使用裁判代替

较少，更丰富的节点

一个常见错误是将工作分成太多小型单一目的节点。每个节点边界都需要序列化输出，丢失上下文信息，并增加边缘复杂性。

坏的（8个薄节点）	好的（4个丰富节点）
parse-query	intake (client-facing)
search-sources	research (search + fetch + analyze)
fetch-content	review (client-facing)
evaluate-sources	report (write + deliver)
synthesize-findings
write-report
quality-check
save-report

为什么较少的节点更好：

LLM在单个节点内保留其工作的完整上下文
搜索、获取和分析的研究报告节点保留其对话历史中的所有源材料
较少的边缘意味着更简单的图和较少的故障点
数据工具（save_data/load_data）在单个节点内处理上下文窗口限制

MCP工具 - 正确使用

MCP工具适用于：

test_node — 使用模拟输入验证节点配置
validate_graph — 检查图结构
configure_loop — 设置事件循环参数
create_session — 跟踪会话状态以进行记账

只是不要： 使用MCP作为主要的构建方法或依赖export_graph

错误处理模式

优雅失败与回退

edges = [
    # 成功路径
    EdgeSpec(id="api-success", source="api-call", target="process-results",
             condition=EdgeCondition.ON_SUCCESS),
    # 失败时回退
    EdgeSpec(id="api-to-fallback", source="api-call", target="fallback-cache",
             condition=EdgeCondition.ON_FAILURE, priority=1),
    # 如果回退也失败，则报告
    EdgeSpec(id="fallback-to-error", source="fallback-cache", target="report-error",
             condition=EdgeCondition.ON_FAILURE, priority=1),
]

交接测试

当代理完成时，过渡到测试阶段：

预测试清单

[ ] 代理结构验证：uv run python -m agent_name validate
[ ] 所有节点在nodes/init.py中定义
[ ] 所有边缘连接有效节点，具有正确的优先级
[ ] 反馈边缘目标具有 max_node_visits > 1
[ ] 面向客户端的节点具有有意义的系统提示
[ ] 代理可以导入：from exports.agent_name import default_agent

hive-patterns hive-patterns