hive-patterns

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Building Agents - Patterns & Best Practices

构建Agent - 模式与最佳实践

Design patterns, examples, and best practices for building robust goal-driven agents.

Prerequisites: Complete agent structure using

hive-create

构建稳健的目标驱动型Agent的设计模式、示例及最佳实践。

前置要求： 使用

hive-create

完成Agent结构搭建。

Practical Example: Hybrid Workflow

实战示例：混合工作流

How to build a node using both direct file writes and optional MCP validation:

python

undefined

如何构建同时支持直接文件写入和可选MCP验证的节点：

python

undefined

1. WRITE TO FILE FIRST (Primary - makes it visible)

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. OPTIONALLY VALIDATE WITH MCP (Secondary - bookkeeping)

2. 可选：通过MCP验证（次要步骤 - 记录流程）

validation = mcp__agent-builder__test_node( node_id="search-web", test_input='{"query": "python tutorials"}', mock_llm_response='{"search_results": [...mock results...]}' )


**User experience:**

- Immediately sees node in their editor (from step 1)
- Gets validation feedback (from step 2)
- Can edit the file directly if needed

validation = mcp__agent-builder__test_node( node_id="search-web", test_input='{"query": "python tutorials"}', mock_llm_response='{"search_results": [...mock results...]}' )


**用户体验：**

- 立即在编辑器中看到节点（来自步骤1）
- 获取验证反馈（来自步骤2）
- 必要时可直接编辑文件

Multi-Turn Interaction Patterns

多轮交互模式

For agents needing multi-turn conversations with users, use

client_facing=True

on event_loop nodes.

对于需要与用户进行多轮对话的Agent，可在event_loop节点上设置

client_facing=True

。

Client-Facing Nodes

面向客户端的节点

A client-facing node streams LLM output to the user and blocks for user input between conversational turns. This replaces the old pause/resume pattern.

python

undefined

面向客户端的节点会将LLM输出流式传输给用户，并在对话轮次之间等待用户输入。这替代了旧的暂停/恢复模式。

python

undefined

Client-facing node with STEP 1/STEP 2 prompt pattern

采用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):

Read the topic provided
If it's vague, ask 1-2 clarifying questions
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") """, )

STEP 1 — 读取并回复（仅文本，禁止调用工具）：

读取提供的主题
若主题模糊，提出1-2个澄清问题
若主题明确，确认你的理解

STEP 2 — 用户确认后调用set_output：

set_output("research_brief", "Clear description of what to research") """, )

Internal node runs without user interaction

内部节点无需用户交互即可运行

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"], )


**How it works:**

- Client-facing nodes stream LLM text to the user and block for input after each response
- User input is injected via `node.inject_event(text)`
- When the LLM calls `set_output` to produce structured outputs, the judge evaluates and ACCEPTs
- Internal nodes (non-client-facing) run their entire loop without blocking
- `set_output` is a synthetic tool — a turn with only `set_output` calls (no real tools) triggers user input blocking

**STEP 1/STEP 2 pattern:** Always structure client-facing prompts with explicit phases. STEP 1 is text-only conversation. STEP 2 calls `set_output` after user confirmation. This prevents the LLM from calling `set_output` prematurely before the user responds.


**工作原理：**

- 面向客户端的节点会将LLM文本流式传输给用户，并在每次回复后等待输入
- 用户输入通过`node.inject_event(text)`注入
- 当LLM调用`set_output`生成结构化输出时，判断器会进行评估并ACCEPT
- 内部节点（非面向客户端）会完整运行整个循环而不中断
- `set_output`是一个合成工具 —— 仅包含`set_output`调用的轮次（无真实工具调用）会触发用户输入等待

**STEP 1/STEP 2模式：** 始终为面向客户端的提示设置明确的阶段。STEP 1是纯文本对话，STEP 2在用户确认后调用`set_output`。这可防止LLM在用户回复前过早调用`set_output`。

When to Use client_facing

何时使用client_facing

Scenario	client_facing	Why
Gathering user requirements	Yes	Need user input
Human review/approval checkpoint	Yes	Need human decision
Data processing (scanning, scoring)	No	Runs autonomously
Report generation	No	No user input needed
Final confirmation before action	Yes	Need explicit approval

Legacy Note: The
pause_nodes
/
entry_points
pattern still works for backward compatibility but
client_facing=True
is preferred for new agents.

场景	client_facing	原因
收集用户需求	是	需要用户输入
人工审核/批准检查点	是	需要人工决策
数据处理（扫描、评分）	否	可自主运行
报告生成	否	无需用户输入
操作前最终确认	是	需要明确批准

旧版说明：
pause_nodes
/
entry_points
模式仍可兼容旧版本，但新Agent优先推荐使用
client_facing=True
。

Edge-Based Routing and Feedback Loops

基于边的路由与反馈循环

Conditional Edge Routing

条件边路由

Multiple conditional edges from the same source replace the old

router

node type. Each edge checks a condition on the node's output.

python

undefined

来自同一源的多条条件边替代了旧的

router

节点类型。每条边都会检查节点输出的条件。

python

undefined

Node with mutually exclusive outputs

具有互斥输出的节点

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').", )

Forward edge (positive priority, evaluated first)

正向边（优先级为正，优先评估）

EdgeSpec( id="review-to-campaign", source="review", target="campaign-builder", condition=EdgeCondition.CONDITIONAL, condition_expr="output.get('approved_contacts') is not None", priority=1, )

Feedback edge (negative priority, evaluated after forward edges)

反馈边（优先级为负，在正向边之后评估）

EdgeSpec( id="review-feedback", source="review", target="extractor", condition=EdgeCondition.CONDITIONAL, condition_expr="output.get('redo_extraction') is not None", priority=-1, )


**Key concepts:**

- `nullable_output_keys`: Lists output keys that may remain unset. The node sets exactly one of the mutually exclusive keys per execution.
- `max_node_visits`: Must be >1 on the feedback target (extractor) so it can re-execute. Default is 1.
- `priority`: Positive = forward edge (evaluated first). Negative = feedback edge. The executor tries forward edges first; if none match, falls back to feedback edges.

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`：反馈目标节点（如extractor）的该值必须>1，以便可以重新执行。默认值为1。
- `priority`：正=正向边（优先评估），负=反馈边。执行器会先尝试正向边；若没有匹配的，则回退到反馈边。

Routing Decision Table

路由决策表

Pattern	Old Approach	New Approach
Conditional branching	`router` node	Conditional edges with `condition_expr`
Binary approve/reject	`pause_nodes` + resume	`client_facing=True` + `nullable_output_keys`
Loop-back on rejection	Manual entry_points	Feedback edge with `priority=-1`
Multi-way routing	Router with routes dict	Multiple conditional edges with priorities

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

Judge Patterns

判断器模式

Core Principle: The judge is the SOLE mechanism for acceptance decisions. Never add ad-hoc framework gating to compensate for LLM behavior. If the LLM calls

set_output

prematurely, fix the system prompt or use a custom judge. Anti-patterns to avoid:

Output rollback logic
```
_user_has_responded
```
flags
Premature set_output rejection
Interaction protocol injection into system prompts

Judges control when an event_loop node's loop exits. Choose based on validation needs.

核心原则：判断器是接受决策的唯一机制。 切勿添加临时框架门控来弥补LLM的行为缺陷。如果LLM过早调用

set_output

，请修复系统提示或使用自定义判断器。需避免的反模式：

输出回滚逻辑
```
_user_has_responded
```
标志
过早拒绝set_output
在系统提示中注入交互协议

判断器控制event_loop节点的循环何时退出。可根据验证需求选择合适的判断器。

Implicit Judge (Default)

隐式判断器（默认）

When no judge is configured, the implicit judge ACCEPTs when:

The LLM finishes its response with no tool calls
All required output keys have been set via
```
set_output
```

Best for simple nodes where "all outputs set" is sufficient validation.

未配置判断器时，隐式判断器会在以下情况时ACCEPT：

LLM完成回复且未调用任何工具
所有必填输出键已通过
```
set_output
```
设置

适用于只需“所有输出键已设置”即可完成验证的简单节点。

SchemaJudge

Validates outputs against a Pydantic model. Use when you need structural validation.

python

from pydantic import BaseModel

class ScannerOutput(BaseModel):
    github_users: list[dict]  # Must be a list of user objects

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))

根据Pydantic模型验证输出。适用于需要结构验证的场景。

python

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))

When to Use Which Judge

判断器选择指南

Judge	Use When	Example
Implicit (None)	Output keys are sufficient validation	Simple data extraction
SchemaJudge	Need structural validation of outputs	API response parsing
Custom	Domain-specific validation logic	Score must be 0.0-1.0

判断器	适用场景	示例
隐式（无）	输出键已足够完成验证	简单数据提取
SchemaJudge	需要输出结构验证	API响应解析
自定义	领域特定的验证逻辑	分数必须在0.0-1.0之间

Fan-Out / Fan-In (Parallel Execution)

扇出/扇入（并行执行）

Multiple ON_SUCCESS edges from the same source trigger parallel execution. All branches run concurrently via

asyncio.gather()

python

undefined

来自同一源的多条ON_SUCCESS边会触发并行执行。所有分支会通过

asyncio.gather()

并发运行。

python

undefined

Scanner fans out to Profiler and Scorer in parallel

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)

Both fan in to Extractor

两者都扇入到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)


**Requirements:**

- Parallel event_loop nodes must have **disjoint output_keys** (no key written by both)
- Only one parallel branch may contain a `client_facing` node
- Fan-in node receives outputs from all completed branches in shared memory


**要求：**

- 并行event_loop节点必须具有**不相交的output_keys**（无同时被两个节点写入的键）
- 仅一个并行分支可包含`client_facing`节点
- 扇入节点会在共享内存中接收所有已完成分支的输出

Context Management Patterns

上下文管理模式

Tiered Compaction

分层压缩

EventLoopNode automatically manages context window usage with tiered compaction:

Pruning — Old tool results replaced with compact placeholders (zero-cost, no LLM call)
Normal compaction — LLM summarizes older messages
Aggressive compaction — Keeps only recent messages + summary
Emergency — Hard reset with tool history preservation

EventLoopNode会通过分层压缩自动管理上下文窗口的使用：

修剪 —— 旧工具结果会被替换为紧凑的占位符（零成本，无需LLM调用）
常规压缩 —— LLM对旧消息进行总结
激进压缩 —— 仅保留最近的消息 + 总结
紧急情况 —— 硬重置并保留工具历史

Spillover Pattern

溢出模式

The framework automatically truncates large tool results and saves full content to a spillover directory. The LLM receives a truncation message with instructions to use

load_data

to read the full result.

For explicit data management, use the data tools (real MCP tools, not synthetic):

python

undefined

框架会自动截断大型工具结果，并将完整内容保存到溢出目录。LLM会收到一条截断消息，其中包含使用

load_data

读取完整结果的说明。

如需显式数据管理，请使用数据工具（真实的MCP工具，而非合成工具）：

python

undefined

save_data, load_data, list_data_files, serve_file_to_user are real MCP tools

save_data, load_data, list_data_files, serve_file_to_user是真实的MCP工具

data_dir is auto-injected by the framework — the LLM never sees it

data_dir由框架自动注入 —— LLM永远不会看到它

Saving large results

保存大型结果

save_data(filename="sources.json", data=large_json_string)

Reading with pagination (line-based offset/limit)

分页读取（基于行的偏移量/限制）

load_data(filename="sources.json", offset=0, limit=50)

Listing available files

列出可用文件

list_data_files()

Serving a file to the user as a clickable link

将文件作为可点击链接提供给用户

serve_file_to_user(filename="report.html", label="Research Report")


Add data tools to nodes that handle large tool results:

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

data_dir

is a framework context parameter — auto-injected at call time.

GraphExecutor.execute()

sets it per-execution via

ToolRegistry.set_execution_context(data_dir=...)

(using

contextvars

for concurrency safety), ensuring it matches the session-scoped spillover directory.

serve_file_to_user(filename="report.html", label="Research Report")


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

```python
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

保证并发安全）为每次执行设置该参数，确保其与会话范围的溢出目录匹配。

Anti-Patterns

反模式

What NOT to Do

切勿执行以下操作

Don't rely on
export_graph
— Write files immediately, not at end
Don't hide code in session — Write to files as components are approved
Don't wait to write files — Agent visible from first step
Don't batch everything — Write incrementally, one component at a time
Don't create too many thin nodes — Prefer fewer, richer nodes (see below)
Don't add framework gating for LLM behavior — Fix prompts or use judges instead

不要依赖
export_graph
—— 立即写入文件，而非在最后写入
不要在会话中隐藏代码 —— 组件获批后立即写入文件
不要延迟写入文件 —— Agent从第一步开始就可见
不要批量处理所有内容 —— 增量写入，一次一个组件
不要创建过多单薄节点 —— 优先选择数量更少、功能更丰富的节点（见下文）
不要为LLM行为添加框架门控 —— 改为修复提示或使用判断器

Fewer, Richer Nodes

更少、更丰富的节点

A common mistake is splitting work into too many small single-purpose nodes. Each node boundary requires serializing outputs, losing in-context information, and adding edge complexity.

Bad (8 thin nodes)	Good (4 rich nodes)
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

Why fewer nodes are better:

The LLM retains full context of its work within a single node
A research node that searches, fetches, and analyzes keeps all source material in its conversation history
Fewer edges means simpler graph and fewer failure points
Data tools (
```
save_data
```
/
```
load_data
```
) handle context window limits within a single node

常见错误是将工作拆分为过多小型单用途节点。每个节点边界都需要序列化输出、丢失上下文信息，并增加边的复杂度。

错误示例（8个单薄节点）	正确示例（4个丰富节点）
parse-query	intake（面向客户端）
search-sources	research（搜索 + 获取 + 分析）
fetch-content	review（面向客户端）
evaluate-sources	report（写入 + 交付）
synthesize-findings
write-report
quality-check
save-report

为什么更少的节点更好：

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

MCP Tools - Correct Usage

MCP工具 - 正确用法

MCP tools OK for:

```
test_node
```
— Validate node configuration with mock inputs
```
validate_graph
```
— Check graph structure
```
configure_loop
```
— Set event loop parameters
```
create_session
```
— Track session state for bookkeeping

Just don't: Use MCP as the primary construction method or rely on export_graph

MCP工具适用于：

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

切勿： 将MCP作为主要构建方法或依赖export_graph

Error Handling Patterns

错误处理模式

Graceful Failure with Fallback

带回退的优雅失败

python

edges = [
    # Success path
    EdgeSpec(id="api-success", source="api-call", target="process-results",
             condition=EdgeCondition.ON_SUCCESS),
    # Fallback on failure
    EdgeSpec(id="api-to-fallback", source="api-call", target="fallback-cache",
             condition=EdgeCondition.ON_FAILURE, priority=1),
    # Report if fallback also fails
    EdgeSpec(id="fallback-to-error", source="fallback-cache", target="report-error",
             condition=EdgeCondition.ON_FAILURE, priority=1),
]

python

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),
]

Handoff to Testing

移交至测试阶段

When agent is complete, transition to testing phase:

Agent完成后，过渡到测试阶段：

Pre-Testing Checklist

测试前检查清单

Agent structure validates:
```
uv run python -m agent_name validate
```
All nodes defined in nodes/init.py
All edges connect valid nodes with correct priorities
Feedback edge targets have
```
max_node_visits > 1
```
Client-facing nodes have meaningful system prompts

Agent can be imported:

from exports.agent_name import default_agent

Agent结构验证通过：
```
uv run python -m agent_name validate
```
所有节点已定义在nodes/init.py中
所有边连接有效节点且优先级正确
反馈边的目标节点
```
max_node_visits > 1
```
面向客户端的节点具有有意义的系统提示

Agent可被导入：

from exports.agent_name import default_agent