name: hive-concepts description: 目标驱动智能体的核心概念 - 架构、节点类型（event_loop、function）、工具发现和工作流概述。在开始智能体开发或需要理解智能体基础知识时使用。 license: Apache-2.0 metadata: author: hive version: “2.0” type: foundational part_of: hive

构建智能体 - 核心概念

构建目标驱动智能体（作为Python包）的基础知识。

架构：Python服务（非JSON配置）

智能体作为Python包构建：

exports/my_agent/
├── __init__.py          # 包导出
├── __main__.py          # CLI（运行、信息、验证、shell）
├── agent.py             # 图构建（目标、边、智能体类）
├── nodes/__init__.py    # 节点定义（NodeSpec）
├── config.py            # 运行时配置
└── README.md            # 文档

关键原则：智能体在构建过程中可见且可编辑

组件批准后立即创建文件
用户可以在编辑器中观看文件增长
无会话状态 - 仅直接文件写入
无“导出”步骤 - 构建完成时智能体即准备就绪

核心概念

目标

成功标准和约束（写入agent.py）

goal = Goal(
    id="research-goal",
    name="技术研究智能体",
    description="深入研究技术主题",
    success_criteria=[
        SuccessCriterion(
            id="completeness",
            description="涵盖主题的所有方面",
            metric="coverage_score",
            target=">=0.9",
            weight=0.4,
        ),
        # 总共3-5个成功标准
    ],
    constraints=[
        Constraint(
            id="accuracy",
            description="所有信息必须经过验证",
            constraint_type="hard",
            category="quality",
        ),
        # 总共1-5个约束
    ],
)

节点

工作单元（写入nodes/init.py）

节点类型：

event_loop — 具有工具执行和基于评判器评估的多轮流式循环。可与工具一起使用或不使用工具。
function — 确定性Python操作。不涉及LLM。

search_node = NodeSpec(
    id="search-web",
    name="搜索网络",
    description="搜索信息并提取结果",
    node_type="event_loop",
    input_keys=["query"],
    output_keys=["search_results"],
    system_prompt="搜索网络以查找：{query}。使用web_search工具查找结果，然后调用set_output存储它们。",
    tools=["web_search"],
)

事件循环节点的NodeSpec字段：

字段	默认值	描述
`client_facing`	`False`	如果为True，将输出流式传输给用户，并在轮次之间阻塞等待输入
`nullable_output_keys`	`[]`	可能保持未设置的输出键（用于互斥输出）
`max_node_visits`	`1`	每次运行此节点执行的最大次数。对于反馈循环目标设置>1

边

节点之间的连接（写入agent.py）

边条件：

on_success — 如果节点成功则继续（最常见）
on_failure — 处理错误
always — 始终继续
conditional — 基于评估节点输出的表达式

边优先级：

优先级控制当多条边离开同一节点时的评估顺序。优先级较高的边首先评估。使用负优先级用于反馈边（循环回较早节点的边）。

# 前向边（首先评估）
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,
)

面向客户端的节点

对于与用户的多轮对话，在节点上设置client_facing=True。该节点将：

将其LLM输出直接流式传输给最终用户
在对话轮次之间阻塞等待用户输入
当通过inject_event()注入新输入时恢复

intake_node = NodeSpec(
    id="intake",
    name="需求收集",
    description="从用户处收集需求",
    node_type="event_loop",
    client_facing=True,
    input_keys=[],
    output_keys=["repo_url", "project_url"],
    system_prompt="您是需求收集智能体。向用户询问仓库URL和项目URL。",
)

遗留说明： 旧的pause_nodes / entry_points模式仍然有效，但对于新智能体，client_facing=True是首选。

步骤1 / 步骤2提示模式： 对于面向客户端的节点，使用两个明确阶段构建系统提示：

system_prompt="""\
**步骤1 — 响应用户（仅文本，无工具调用）：**
[呈现信息、提问等]

**步骤2 — 用户响应后，调用set_output：**
[使用结构化输出调用set_output]
"""

这可以防止LLM在用户有机会响应之前过早调用set_output。

节点设计：更少、更丰富的节点

优先选择做更多工作的更少节点，而不是许多单一用途的薄节点：

不好：8个薄节点（解析查询 → 搜索 → 获取 → 评估 → 合成 → 写入 → 检查 → 保存）
好：4个丰富节点（需求收集 → 研究 → 审查 → 报告）

原因：每个节点边界都需要序列化输出和传递上下文。更少的节点意味着LLM在节点内保留其工作的完整上下文。一个进行搜索、获取和分析的研究节点在其对话历史中保留所有源材料。

用于跨边输入的nullable_output_keys

当节点接收仅在某些边上到达的输入时（例如，feedback仅来自审查 → 研究反馈循环，而不是来自需求收集 → 研究），将这些键标记为nullable_output_keys：

research_node = NodeSpec(
    id="research",
    input_keys=["research_brief", "feedback"],
    nullable_output_keys=["feedback"],  # 首次访问时不存在
    max_node_visits=3,
    ...
)

事件循环架构概念

EventLoopNode如何工作

事件循环节点运行多轮循环：

LLM接收系统提示 + 对话历史
LLM响应（文本和/或工具调用）
工具调用被执行，结果添加到对话中
评判器评估：接受（退出循环）、重试（再次循环）或升级
重复直到评判器接受或达到max_iterations

EventLoopNode运行时

EventLoopNodes在运行时由GraphExecutor自动创建。您不需要手动注册它们。GraphExecutor（直接）和AgentRuntime / create_agent_runtime()都自动处理event_loop节点。

# 直接执行 — 执行器自动创建EventLoopNodes
from framework.graph.executor import GraphExecutor
from framework.runtime.core import Runtime

runtime = Runtime(storage_path)
executor = GraphExecutor(
    runtime=runtime,
    llm=llm,
    tools=tools,
    tool_executor=tool_executor,
    storage_path=storage_path,
)
result = await executor.execute(graph=graph, goal=goal, input_data=input_data)

# TUI执行 — AgentRuntime也有效
from framework.runtime.agent_runtime import create_agent_runtime
runtime = create_agent_runtime(
    graph=graph, goal=goal, storage_path=storage_path,
    entry_points=[...], llm=llm, tools=tools, tool_executor=tool_executor,
)

set_output

节点通过调用set_output(key, value)产生结构化输出 — 这是框架注入的合成工具。当LLM调用set_output时，值存储在输出累加器中，并通过共享内存提供给下游节点。

set_output不是真正的工具 — 它被排除在real_tool_results之外。对于面向客户端的节点，这意味着LLM仅调用set_output（无其他工具）的轮次被视为对话边界，并将阻塞等待用户输入。

JudgeProtocol

评判器是接受决策的唯一机制。 不要添加临时框架门控、输出回滚或过早拒绝逻辑。如果LLM过早调用set_output，请通过更好的提示或自定义评判器修复 — 而不是框架级防护。

评判器控制节点的循环何时退出：

隐式评判器（默认，未配置评判器）：当LLM完成且无工具调用且所有必需输出键都已设置时接受
SchemaJudge：根据Pydantic模型验证输出
自定义评判器：实现evaluate(context) -> JudgeVerdict

LoopConfig

控制循环行为：

max_iterations（默认50） — 防止无限循环
max_tool_calls_per_turn（默认10） — 限制每个LLM响应的工具调用
tool_call_overflow_margin（默认0.5） — 在丢弃额外工具调用之前的余量（50%意味着硬截止点为限制的150%）
stall_detection_threshold（默认3） — 检测重复的相同响应
max_history_tokens（默认32000） — 触发对话压缩

数据工具（溢出管理）

当工具结果超出上下文窗口时，框架自动将它们保存到溢出目录并进行截断提示。产生或消费大量数据的节点应包含数据工具：

save_data(filename, data) — 将数据写入数据目录中的文件
load_data(filename, offset=0, limit=50) — 使用基于行的分页读取数据
list_data_files() — 列出可用的数据文件
serve_file_to_user(filename, label="") — 为用户获取可点击的file:// URI

注意：data_dir是框架注入的上下文参数 — LLM永远不会看到或传递它。GraphExecutor.execute()通过contextvars每次执行设置它，因此数据工具和溢出始终共享相同的会话范围目录。

这些是真正的MCP工具（非合成）。将它们添加到处理大型工具结果的节点中：

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

扇出/扇入

来自同一源的多条ON_SUCCESS边创建并行执行。所有分支通过asyncio.gather()并发运行。并行event_loop节点必须具有不相交的output_keys。

max_node_visits

控制节点在一次图运行中可以执行的次数。默认值为1。对于作为反馈边目标的节点（审查-拒绝循环）设置更高。设置为0表示无限（由max_steps保护）。

工具发现与验证

关键： 在添加带有工具的节点之前，您必须验证工具是否存在。

工具由MCP服务器提供。切勿假设工具存在 — 始终动态发现。

步骤1：注册MCP服务器（如果尚未完成）

mcp__agent-builder__add_mcp_server(
    name="tools",
    transport="stdio",
    command="python",
    args='["mcp_server.py", "--stdio"]',
    cwd="../tools"
)

步骤2：发现可用工具

# 列出所有已注册服务器的所有工具
mcp__agent-builder__list_mcp_tools()

# 或列出特定服务器的工具
mcp__agent-builder__list_mcp_tools(server_name="tools")

步骤3：添加节点前验证

在编写带有tools=[...]的节点之前：

调用list_mcp_tools()获取可用工具
检查节点中的每个工具是否存在于响应中
如果工具不存在：
- 不要继续该节点
- 通知用户：“工具’X’不可用。可用工具有：…”
- 询问他们是否想使用替代方案或继续不使用该工具

工具验证反模式

切勿假设工具存在 — 始终首先调用list_mcp_tools()
切勿编写带有未验证工具的节点 — 在编写前验证
切勿静默删除工具 — 如果工具不存在，通知用户
切勿猜测工具名称 — 使用发现响应中的确切名称

工作流概述：增量文件构建

1. 创建包 → mkdir + 编写骨架
2. 定义目标 → 写入agent.py + config.py
3. 对于每个节点：
   - 提出设计（event_loop用于LLM工作，function用于确定性）
   - 用户批准
   - 立即写入nodes/__init__.py
   - （可选）使用test_node验证
4. 连接边 → 更新agent.py
   - 使用优先级用于反馈边（负优先级）
   - （可选）使用validate_graph验证
5. 完成 → 将智能体类写入agent.py
6. 完成 - 智能体准备就绪于exports/my_agent/

文件立即写入。MCP工具可选用于验证/测试簿记。

何时使用此技能

在以下情况下使用hive-concepts：

开始新的智能体项目并需要理解基础知识
在构建之前需要理解智能体架构
想要在继续之前验证工具可用性
学习节点类型、边和图执行

后续步骤：

准备构建？ → 使用hive-create技能
需要模式和示例？ → 使用hive-patterns技能

用于验证的MCP工具

写入文件后，可选地使用MCP工具进行验证：

test_node - 使用模拟输入验证节点配置

mcp__agent-builder__test_node(
    node_id="search-web",
    test_input='{"query": "test query"}',
    mock_llm_response='{"results": "mock output"}'
)

validate_graph - 检查图结构

mcp__agent-builder__validate_graph()
# 返回：不可达节点、缺失连接、event_loop验证等。

configure_loop - 设置事件循环参数

mcp__agent-builder__configure_loop(
    max_iterations=50,
    max_tool_calls_per_turn=10,
    stall_detection_threshold=3,
    max_history_tokens=32000
)

关键点： 文件首先写入。MCP工具仅用于验证。

蜂巢智能体构建核心概念Skill hive-concepts