人工智能AI Agent多智能体MCP 服务工具调用浏览器控制【免费下载链接】hiveMulti-Agent Harness for Production AI项目地址https://gitcode.com/gh_mirrors/hive48/hive点击查看免费下载导读本指南围绕 HiveMulti-Agent Harness for Production AI的架构设计文档《Multi-Entry-Point Agent Architecture》展开讲解如何让一个 Agent 同时处理 Webhook、API 请求、定时器事件与内部事件等多类异步入口以及为什么用工具当共享内存是必须规避的反模式。读完本文你将掌握 Hive 中AgentHost多入口运行时、ExecutionManager执行流、隔离级别IsolationLevel与事件总线EventBus的完整设计脉络与实战配置方法并理解从旧架构迁移到显式状态管理的具体步骤。一、为什么生产环境中的 Agent 需要多个入口点以文档中的 Tier-1 客服支持 Agent 为例它必须同时承担四类异步职责监听 Zendesk Webhook——新工单随时异步到达处理 API 请求——用户可以查询工单状态或提交后续操作处理定时器事件——每 5 分钟执行一次升级检查escalation check响应内部事件——其他 Agent 可能委派任务给它。这些操作不是顺序执行的而是并发且相互独立的Webhook 触发时可能正在处理某个 API 请求两张工单也可能同时到达。这意味着 Agent 运行时必须支持多入口点 并发执行否则任何单一阻塞都会拖垮整条业务链路。旧架构的根本限制早期框架只有一个_current_run字段同一时刻只能保存一次运行# 早期 Runtime 的核心约束文档中 core.py:58 的示意代码 class Runtime: def __init__(self, ...): self._current_run: Run | None None # 同一时刻只能有一个 Run这个单一_current_run带来三个致命问题无法并发执行——处理一张工单会阻塞其他所有工单只有一个入口点——只有entry_node能启动执行状态互相覆盖——并发的尝试会互相覆盖彼此的上下文。从当前仓库源码看这一约束的演进脉络清晰可循文档中提到的Runtime已被重构为 core/framework/host/agent_host.py 中的AgentHost仓库中实际不存在core/framework/runtime/目录实现全部收敛到core/framework/host/它以streams: dict[str, ExecutionManager]按入口点 ID 管理多条执行流彻底取代了单一运行模型。二、为什么Tools-as-Shared-Memory是反模式一个看似省事的变通方案是用工具Tool去读写外部存储从而在多次执行之间共享状态# 反模式用工具管理共享状态 tool def get_customer_context(customer_id: str) - dict: 从数据库检索客户上下文。 return db.get_customer(customer_id) tool def update_ticket_status(ticket_id: str, status: str) - bool: 更新数据库中的工单状态。 db.update_ticket(ticket_id, status) return True表面上这能工作——工具可以读写外部存储形成执行间的共享状态。但文档明确指出这种方法在并发生产场景下存在五个严重问题。1. 无隔离控制的竞态条件执行 A: get_customer_context(cust_123) → {tickets: 5} 执行 B: get_customer_context(cust_123) → {tickets: 5} 执行 A: update_ticket_count(cust_123, 6) 执行 B: update_ticket_count(cust_123, 6) # 应该是 7工具没有隔离级别isolation level的概念每次调用都直接访问存储、互不协调。高并发下必然出现丢失更新Lost updates——后写覆盖先写脏读Dirty reads——读到写了一半的状态幻读Phantom data——同一逻辑操作内状态发生变化。2. 没有事务边界工具独立执行不具备任何事务语义# 如果这个流程中途失败会怎样 tool def process_refund(order_id: str) - dict: mark_order_refunded(order_id) # ✓ 成功 credit_customer_account(order_id) # ✗ 失败——网络错误 send_confirmation_email(order_id) # 永远不会执行 # 现在订单标记为已退款但客户并没有收到退款工具即状态时无法做到回滚部分变更保证操作原子性协调多步状态转换。3. 隐形依赖破坏目标评估Hive 的目标驱动goal-driven机制依赖决策—结果的追踪# 决策根据购买历史升级客户等级 # 结果成功/失败且伴有可观测的状态变化当状态通过工具流动时框架对内部过程完全失去可见性tool def update_customer_tier(customer_id: str) - str: # 它读了什么状态改了哪些状态 # 框架完全不知道——它只看到工具返回 gold history get_purchase_history(customer_id) # 隐藏的读 new_tier calculate_tier(history) # 隐藏的逻辑 save_tier(customer_id, new_tier) # 隐藏的写 return new_tier这会破坏结果聚合Outcome aggregation——无法追踪跨执行的状态变更约束检查Constraint checking——无法验证不变量是否被保持目标进度评估Goal progress evaluation——无法把动作与成功标准关联起来。4. 没有执行关联能力当多个入口点并发触发时你需要追踪哪次执行修改了哪份状态关联相关操作例如同一张工单的 webhook 与后续 API 调用通过追踪执行流来排查问题。工具调用彼此完全独立、不携带任何执行上下文以上能力一概没有。5. 测试变得不可能工具即状态时单元测试无法隔离状态——每个测试都会污染全局存储并发测试互相干扰Mock需要替换真实的数据库/API 调用。对比真正的显式状态管理# 隔离的测试——无任何外部依赖 buf manager.create_buffer(test-exec, test-stream, IsolationLevel.ISOLATED) await buf.write(key, value) assert await buf.read(key) value # 其他测试不受影响三、解决方案显式状态管理架构新架构引入显式状态管理Explicit State Management用显式的隔离与追踪取代隐式的工具副作用┌─────────────────────────────────────────────────────┐ │ AgentRuntime │ │ - 管理 Agent 生命周期 │ │ - 协调 ExecutionStreams │ │ - 聚合结果以支持目标评估 │ ├─────────────────────────────────────────────────────┤ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Stream A │ │ Stream B │ │ Stream C │ │ │ │ (webhook) │ │ (api) │ │ (timer) │ │ │ │ │ │ │ │ │ │ │ │ Concurrent │ │ Concurrent │ │ Concurrent │ │ │ │ Executions │ │ Executions │ │ Executions │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ └────────────────┼────────────────┘ │ │ ↓ │ │ SharedBufferManager │ │ (Isolation Levels) │ │ │ │ OutcomeAggregator │ │ (Cross-Stream Goals) │ └─────────────────────────────────────────────────────┘说明上图为文档中的原始架构示意。需要特别指出的是当前仓库经过 Colony 重构后shared_state.py与outcome_aggregator.py已被精简为兼容桩stub——见下文实现现状一节其状态共享职责实际收敛到会话存储SessionStore与state.json层。核心组件 1SharedBufferManager 与隔离级别class IsolationLevel(Enum): ISOLATED isolated # 每次执行私有状态 SHARED shared # 跨执行可见最终一致性 SYNCHRONIZED synchronized # 共享且带写锁强一致性每次执行都可以显式控制状态可见性# 执行级状态不受干扰安全 await memory.write(scratch_data, value, scopeStateScope.EXECUTION) # 流级共享状态同一流内所有执行可见 await memory.write(stream_counter, count, scopeStateScope.STREAM) # 全局状态处处可见谨慎使用 await memory.write(system_config, config, scopeStateScope.GLOBAL)实现现状仓库佐证core/framework/host/shared_state.py 中定义了IsolationLevelisolated/shared/synchronized与StateScopeexecution/stream/global两个枚举以及SharedBufferManager的create_buffer/get_stream_state/get_global_state/cleanup_execution等接口。文件头部注释明确标注Stub — shared state removed in colony refactor当前实现中共享缓冲区仅作为兼容层存在get_recent_changes恒返回空列表真正的跨执行状态持久化由SessionStore的state.json与resume_session_id机制承载——这一演进在 execution_manager.py 的_run_execution中可见共享会话的执行会跳过初始/最终状态写入由主执行通过_write_progress()统一维护state.json。核心组件 2StreamRuntime 与执行追踪class StreamRuntime: def __init__(self, stream_id, storage, outcome_aggregator): # 按 execution_id 追踪运行而非单一 _current_run self._runs: dict[str, Run] {}现在多个执行可以并发运行而不冲突# 执行 A runtime.start_run(execution_idexec-A, goal_idsupport) runtime.decide(execution_idexec-A, intentclassify ticket, ...) # 执行 B并发无冲突 runtime.start_run(execution_idexec-B, goal_idsupport) runtime.decide(execution_idexec-B, intentclassify ticket, ...)实现现状仓库佐证仓库中的对应实现为 core/framework/host/stream_runtime.py 的StreamDecisionTracker——文档描述其关键差异正是通过execution_id并发追踪多个 run以及使用ConcurrentStorage保证线程安全的持久化。而每个入口点的执行编排由 core/framework/host/execution_manager.py 的ExecutionManager承担它持有_active_executionsExecutionContext字典、_execution_tasksasyncio.Task、_execution_results带 TTL 与数量上限的OrderedDict等簿记结构并内置复活机制resurrection——非致命错误下从失败节点自动重启默认最多max_resurrections3次_FATAL_ERROR_PATTERNS中列出的凭据/鉴权/配置类错误则直接判定致命、不复活。核心组件 3OutcomeAggregator 与跨流目标class OutcomeAggregator: def record_decision(self, stream_id, execution_id, decision) - None def record_outcome(self, stream_id, execution_id, decision_id, outcome) - None async def evaluate_goal_progress(self) - dict框架现在追踪所有流上的全部决策从而支持统一的目标进度评估跨执行的约束违规检测带正确归因的成功标准追踪。实现现状仓库佐证core/framework/host/outcome_aggregator.py 目前为兼容桩evaluate_goal_progress()返回占位结果AgentHost.get_goal_progress()见 agent_host.py会调用该聚合器完成跨流目标评估聚合器内部还接入EventBus以便发布GOAL_PROGRESS/GOAL_ACHIEVED/CONSTRAINT_VIOLATION等事件。核心组件 4EventBus 事件协调# 流 A 发布 await bus.publish(AgentEvent( typeEventType.EXECUTION_COMPLETED, stream_idwebhook, execution_idexec-123, data{ticket_resolved: True}, )) # 流 B 订阅 bus.subscribe( event_types[EventType.EXECUTION_COMPLETED], handleron_ticket_resolved, filter_streamwebhook, )流之间无需紧耦合或共享可变状态即可协调。实现现状仓库佐证core/framework/host/event_bus.py 定义了EventTypeStrEnum枚举覆盖执行生命周期EXECUTION_STARTED/EXECUTION_COMPLETED/EXECUTION_FAILED/EXECUTION_PAUSED/EXECUTION_RESUMED、状态变更STATE_CHANGED/STATE_CONFLICT、目标追踪GOAL_PROGRESS/GOAL_ACHIEVED/CONSTRAINT_VIOLATION、流生命周期、LLM 流式可观测性、工具调用、外部触发WEBHOOK_RECEIVED以及 Worker 生命周期WORKER_COMPLETED/WORKER_FAILED等 60 种事件类型。subscribe()支持filter_stream/filter_node/filter_graph三类过滤器AgentHost还提供subscribe_to_events()/unsubscribe_from_events()包装方法agent_host.py。另外execution_manager.py 中的GraphScopedEventBus会在每次publish()时自动给事件打上graph_id戳并代理订阅、历史与统计接口。四、什么时候该用 Tool决策对照表工具仍然是以下场景的正确选择外部系统集成——调用 API、数据库、服务副作用——发送邮件、创建资源数据检索——获取决策所需的信息。关键区分在于职责边界使用场景正确做法在多次执行之间协调SharedBufferManager追踪决策结果StreamRuntime OutcomeAggregator调用外部 APITool持久化业务数据Tool写入外部存储执行期间共享草稿状态StreamBuffer向其他流发布事件EventBus五、实战配置在 AgentHost 中注册多入口点当前仓库中AgentHost即文档所称AgentRuntime的现行实现通过EntryPointSpec注册每个入口点。以下综合 agent_host.py 的AgentRuntimeConfig与 execution_manager.py 的EntryPointSpec给出完整可运行示例runtime AgentHost( graphsupport_agent_graph, goalsupport_agent_goal, storage_pathPath(./storage), llmllm_provider, ) # 注册 Webhook 入口 runtime.register_entry_point(EntryPointSpec( idwebhook, nameZendesk Webhook, entry_nodeprocess-webhook, trigger_typewebhook, isolation_levelshared, )) # 注册 API 入口 runtime.register_entry_point(EntryPointSpec( idapi, nameAPI Handler, entry_nodeprocess-request, trigger_typeapi, isolation_levelshared, )) # 注册定时器入口每 5 分钟一次升级检查 runtime.register_entry_point(EntryPointSpec( idtimer, nameEscalation Check, entry_nodecheck-escalations, trigger_typetimer, trigger_config{interval_minutes: 5}, isolation_levelisolated, )) # 注册事件驱动入口监听其他流的完成事件 runtime.register_entry_point(EntryPointSpec( idevents, nameDelegate Responder, entry_nodehandle-delegated-task, trigger_typeevent, trigger_config{event_types: [execution_completed]}, isolation_levelshared, )) # 启动运行时会为每个入口点创建 ExecutionManager 流 await runtime.start() # 非阻塞触发立即返回 execution_id exec_1 await runtime.trigger(webhook, {ticket_id: 123}) exec_2 await runtime.trigger(api, {query: help}) # 检查目标进度 progress await runtime.get_goal_progress() print(fProgress: {progress[overall_progress]:.1%}) # 停止运行时 await runtime.stop()EntryPointSpec 字段详解字段类型默认值说明idstr必填入口点唯一标识同时作为流 IDnamestr必填入口点显示名entry_nodestr必填启动执行的图节点 ID必须存在于图中否则register_entry_point抛ValueErrortrigger_typestr必填webhook/api/timer/event/manualtrigger_configdict{}触发器附加配置见下文isolation_levelstrsharedisolated/shared/synchronized对应IsolationLevel枚举priorityint0优先级max_concurrentint10该入口点允许的最大并发执行数对应ExecutionManager内部的asyncio.Semaphoremax_resurrectionsint3非致命失败时的自动重启次数0 表示禁用AgentRuntimeConfig 关键参数agent_host.py 中的AgentRuntimeConfig提供如下可调项参数默认值说明max_concurrent_executions100全局最大并发执行数cache_ttl/batch_interval60.0/0.1传给ConcurrentStorage的缓存 TTL 与批量写入间隔max_history1000事件总线保留的历史事件数execution_result_max1000每个流保留的执行结果上限超出按 FIFO 淘汰execution_result_ttl_secondsNone执行结果保留 TTLNone 表示不过期idempotency_ttl_seconds/idempotency_max_keys300.0/10000trigger()幂等去重窗口与最大键数webhook_host/webhook_port127.0.0.1/8080Webhook 服务器监听地址与端口webhook_routes[]Webhook 路由列表非空才会启动服务器各触发器类型的 trigger_config 配置从AgentHost.start()与_start_timers()的实现agent_host.py可以归纳timer固定间隔模式{interval_minutes: 5, run_immediately: false, idle_timeout_seconds: 300}timerCron 表达式模式{cron: */5 * * * *, run_immediately: false, idle_timeout_seconds: 300}需要安装croniter缺失时启动抛RuntimeError并提示uv pip install croniterevent{event_types: [execution_completed, execution_failed], filter_stream: webhook, filter_node: ..., filter_graph: ..., exclude_own_graph: false}webhook对应AgentRuntimeConfig.webhook_routes每个路由形如{source_id: str, path: str, methods: [POST], secret: str|None}由 core/framework/host/webhook_server.py 的WebhookServer承载收到请求后会发布WEBHOOK_RECEIVED事件。仓库中自带一个真实的 timer 触发器示例见 examples/templates/email_inbox_management/triggers.json[ { id: email-timer, name: Scheduled Inbox Check, trigger_type: timer, trigger_config: { interval_minutes: 5 }, task: Fetch and process inbox emails according to the users rules } ]定时器的避让门控逻辑值得注意的一个工程细节AgentHost的定时器循环并非机械地到点触发而是带有避让门控——若任何流的执行正在活跃工作LLM/工具调用未超过idle_timeout_seconds默认 300 秒则跳过本次 tick避免定时任务打断正在进行的会话pause_timers()/resume_timers()可显式暂停/恢复所有定时器。这套逻辑在 agent_host.py 的_start_timers中实现并且在trigger()时会先经过 Pipeline 中间件限流、校验、成本防护等被拒时抛PipelineRejectedError。幂等触发与等待结果针对 Webhook 提供方超时重试的场景trigger()支持幂等键去重agent_host.py# 第一次触发缓存 execution_id exec_id await runtime.trigger(webhook, {ticket_id: 123}, idempotency_keywh-2026-0001) # 重试到达TTL 窗口内返回相同 execution_id不重复执行 exec_id_again await runtime.trigger(webhook, {ticket_id: 123}, idempotency_keywh-2026-0001) assert exec_id exec_id_again需要同步等待完成时使用trigger_and_wait(entry_point_id, input_data, timeout...)返回ExecutionResult或超时返回Noneagent_host.py。会话共享异步入口如何加入主会话AgentHost._get_primary_session_state()agent_host.py实现了关键的会话协作当 Webhook/定时器等异步入口以isolation_levelshared触发时会查找主会话如用户前端会话的活跃执行返回{resume_session_id: exec_id, data_buffer: {...}}从而复用同一个会话目录——记忆如用户定义的规则共享、日志落在同一会话目录且data_buffer只注入异步入口节点声明为输入的键避免上一轮过期的输出泄漏进来。而isolation_levelisolated的入口则使用独立会话目录并持久复用同一会话保证conversation_modecontinuous生效。多图扩展AgentHost还支持在运行中动态加载次级图add_graph()agent_host.py每个次级图获得独立的SessionStore与RuntimeLogStore会话/日志写入graphs/{graph_id}子目录共享同一个EventBus、state.json与数据目录并同样支持事件订阅与定时器入口。list_graphs()/remove_graph()/active_graph_id提供管理能力——这可以看作是多入口点在多 Agent图维度上的自然延伸。六、迁移指南从反模式到正确架构Before反模式# tools.py - 状态隐藏在工具里 tool def get_processing_count() - int: return redis.get(processing_count) or 0 tool def increment_processing_count() - int: return redis.incr(processing_count)After正确架构# 在节点执行中 async def execute(self, context, memory): # 从受管状态读取 count await memory.read(processing_count) or 0 # 带显式隔离级别更新 await memory.write( processing_count, count 1, scopeStateScope.STREAM, # 显式作用域 )迁移检查清单盘点现状找出所有在工具内部读写全局状态的代码Redis 计数器、内存字典、环境变量写等分类职责业务数据持久化保留在工具写入外部存储执行协调状态迁移到受管状态memory.read/writeStateScope流间通知改为EventBus发布/订阅设定隔离级别每个入口点在EntryPointSpec.isolation_level中显式声明isolated/shared/synchronized验证目标评估迁移后确认get_goal_progress()能正确归因跨流决策STATE_CONFLICT/CONSTRAINT_VIOLATION事件可按预期触发补充并发测试为每个入口点编写并发触发测试确认互不干扰、执行结果可独立追踪。七、总结两种方案对比方面Tools-as-State显式状态管理并发竞态条件隔离级别事务无执行级作用域可见性隐藏可观测测试需要 Mock设计上天然隔离目标追踪被破坏完整归因调试不透明可追踪多入口点架构的意义不止于让并发执行成为可能——它为可靠、可观测、目标驱动的 Agent 提供了地基让 Agent 能在生产环境中安全运行。在 Hive 当前代码库中这套设计的落地载体是 core/framework/host/agent_host.py生命周期与入口点管理、core/framework/host/execution_manager.py执行流与并发控制、core/framework/host/stream_runtime.py按执行 ID 的决策追踪、core/framework/host/event_bus.py事件协调以及 core/framework/host/webhook_server.py外部 HTTP 入口。相关行为测试可参考 core/tests/test_trigger_fires_into_queen.py、core/tests/test_node_conversation.py 与 core/tests/test_progress_db.py。需要提醒的是文档中所引的core/framework/runtime/目录路径在当前仓库中已不存在以上实现均位于core/framework/host/下阅读源码时请以实际路径为准。赞分享人工智能AI Agent多智能体MCP 服务工具调用浏览器控制【免费下载链接】hiveMulti-Agent Harness for Production AI项目地址https://gitcode.com/gh_mirrors/hive48/hive点击查看免费下载相关推荐Hive Agent Framework 声明式 Agent 构建指南agent.json 架构、节点/边配置与执行图实战Hive Agent Framework 声明式 Agent 构建指南agent.json 架构、节点/边配置与执行图实战 本篇技术指南以 Hive Agen人工智能AI Agent多智能体MCP 服务工具调用浏览器控制从单点到多活Apollo无状态服务架构的99.99%可用性实践从单点到多活Apollo无状态服务架构的99.99%可用性实践 你是否曾因配置中心单点故障导致全网服务不可用是否在机房断电时眼睁睁看着配置更新全部阻塞Ap配置中心后端微服务Agent OS 与 AgentMesh 统一治理架构从单机内核到分布式 Agent 生态的端到端治理实战Agent OS 与 AgentMesh 统一治理架构从单机内核到分布式 Agent 生态的端到端治理实战 Agent OS 与 AgentMesh 是 Ag人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Hive 多入口点 Agent 架构：从单一执行到显式状态管理的并发生产实践
人工智能AI Agent多智能体MCP 服务工具调用浏览器控制【免费下载链接】hiveMulti-Agent Harness for Production AI项目地址https://gitcode.com/gh_mirrors/hive48/hive点击查看免费下载导读本指南围绕 HiveMulti-Agent Harness for Production AI的架构设计文档《Multi-Entry-Point Agent Architecture》展开讲解如何让一个 Agent 同时处理 Webhook、API 请求、定时器事件与内部事件等多类异步入口以及为什么用工具当共享内存是必须规避的反模式。读完本文你将掌握 Hive 中AgentHost多入口运行时、ExecutionManager执行流、隔离级别IsolationLevel与事件总线EventBus的完整设计脉络与实战配置方法并理解从旧架构迁移到显式状态管理的具体步骤。一、为什么生产环境中的 Agent 需要多个入口点以文档中的 Tier-1 客服支持 Agent 为例它必须同时承担四类异步职责监听 Zendesk Webhook——新工单随时异步到达处理 API 请求——用户可以查询工单状态或提交后续操作处理定时器事件——每 5 分钟执行一次升级检查escalation check响应内部事件——其他 Agent 可能委派任务给它。这些操作不是顺序执行的而是并发且相互独立的Webhook 触发时可能正在处理某个 API 请求两张工单也可能同时到达。这意味着 Agent 运行时必须支持多入口点 并发执行否则任何单一阻塞都会拖垮整条业务链路。旧架构的根本限制早期框架只有一个_current_run字段同一时刻只能保存一次运行# 早期 Runtime 的核心约束文档中 core.py:58 的示意代码 class Runtime: def __init__(self, ...): self._current_run: Run | None None # 同一时刻只能有一个 Run这个单一_current_run带来三个致命问题无法并发执行——处理一张工单会阻塞其他所有工单只有一个入口点——只有entry_node能启动执行状态互相覆盖——并发的尝试会互相覆盖彼此的上下文。从当前仓库源码看这一约束的演进脉络清晰可循文档中提到的Runtime已被重构为 core/framework/host/agent_host.py 中的AgentHost仓库中实际不存在core/framework/runtime/目录实现全部收敛到core/framework/host/它以streams: dict[str, ExecutionManager]按入口点 ID 管理多条执行流彻底取代了单一运行模型。二、为什么Tools-as-Shared-Memory是反模式一个看似省事的变通方案是用工具Tool去读写外部存储从而在多次执行之间共享状态# 反模式用工具管理共享状态 tool def get_customer_context(customer_id: str) - dict: 从数据库检索客户上下文。 return db.get_customer(customer_id) tool def update_ticket_status(ticket_id: str, status: str) - bool: 更新数据库中的工单状态。 db.update_ticket(ticket_id, status) return True表面上这能工作——工具可以读写外部存储形成执行间的共享状态。但文档明确指出这种方法在并发生产场景下存在五个严重问题。1. 无隔离控制的竞态条件执行 A: get_customer_context(cust_123) → {tickets: 5} 执行 B: get_customer_context(cust_123) → {tickets: 5} 执行 A: update_ticket_count(cust_123, 6) 执行 B: update_ticket_count(cust_123, 6) # 应该是 7工具没有隔离级别isolation level的概念每次调用都直接访问存储、互不协调。高并发下必然出现丢失更新Lost updates——后写覆盖先写脏读Dirty reads——读到写了一半的状态幻读Phantom data——同一逻辑操作内状态发生变化。2. 没有事务边界工具独立执行不具备任何事务语义# 如果这个流程中途失败会怎样 tool def process_refund(order_id: str) - dict: mark_order_refunded(order_id) # ✓ 成功 credit_customer_account(order_id) # ✗ 失败——网络错误 send_confirmation_email(order_id) # 永远不会执行 # 现在订单标记为已退款但客户并没有收到退款工具即状态时无法做到回滚部分变更保证操作原子性协调多步状态转换。3. 隐形依赖破坏目标评估Hive 的目标驱动goal-driven机制依赖决策—结果的追踪# 决策根据购买历史升级客户等级 # 结果成功/失败且伴有可观测的状态变化当状态通过工具流动时框架对内部过程完全失去可见性tool def update_customer_tier(customer_id: str) - str: # 它读了什么状态改了哪些状态 # 框架完全不知道——它只看到工具返回 gold history get_purchase_history(customer_id) # 隐藏的读 new_tier calculate_tier(history) # 隐藏的逻辑 save_tier(customer_id, new_tier) # 隐藏的写 return new_tier这会破坏结果聚合Outcome aggregation——无法追踪跨执行的状态变更约束检查Constraint checking——无法验证不变量是否被保持目标进度评估Goal progress evaluation——无法把动作与成功标准关联起来。4. 没有执行关联能力当多个入口点并发触发时你需要追踪哪次执行修改了哪份状态关联相关操作例如同一张工单的 webhook 与后续 API 调用通过追踪执行流来排查问题。工具调用彼此完全独立、不携带任何执行上下文以上能力一概没有。5. 测试变得不可能工具即状态时单元测试无法隔离状态——每个测试都会污染全局存储并发测试互相干扰Mock需要替换真实的数据库/API 调用。对比真正的显式状态管理# 隔离的测试——无任何外部依赖 buf manager.create_buffer(test-exec, test-stream, IsolationLevel.ISOLATED) await buf.write(key, value) assert await buf.read(key) value # 其他测试不受影响三、解决方案显式状态管理架构新架构引入显式状态管理Explicit State Management用显式的隔离与追踪取代隐式的工具副作用┌─────────────────────────────────────────────────────┐ │ AgentRuntime │ │ - 管理 Agent 生命周期 │ │ - 协调 ExecutionStreams │ │ - 聚合结果以支持目标评估 │ ├─────────────────────────────────────────────────────┤ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ Stream A │ │ Stream B │ │ Stream C │ │ │ │ (webhook) │ │ (api) │ │ (timer) │ │ │ │ │ │ │ │ │ │ │ │ Concurrent │ │ Concurrent │ │ Concurrent │ │ │ │ Executions │ │ Executions │ │ Executions │ │ │ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │ │ └────────────────┼────────────────┘ │ │ ↓ │ │ SharedBufferManager │ │ (Isolation Levels) │ │ │ │ OutcomeAggregator │ │ (Cross-Stream Goals) │ └─────────────────────────────────────────────────────┘说明上图为文档中的原始架构示意。需要特别指出的是当前仓库经过 Colony 重构后shared_state.py与outcome_aggregator.py已被精简为兼容桩stub——见下文实现现状一节其状态共享职责实际收敛到会话存储SessionStore与state.json层。核心组件 1SharedBufferManager 与隔离级别class IsolationLevel(Enum): ISOLATED isolated # 每次执行私有状态 SHARED shared # 跨执行可见最终一致性 SYNCHRONIZED synchronized # 共享且带写锁强一致性每次执行都可以显式控制状态可见性# 执行级状态不受干扰安全 await memory.write(scratch_data, value, scopeStateScope.EXECUTION) # 流级共享状态同一流内所有执行可见 await memory.write(stream_counter, count, scopeStateScope.STREAM) # 全局状态处处可见谨慎使用 await memory.write(system_config, config, scopeStateScope.GLOBAL)实现现状仓库佐证core/framework/host/shared_state.py 中定义了IsolationLevelisolated/shared/synchronized与StateScopeexecution/stream/global两个枚举以及SharedBufferManager的create_buffer/get_stream_state/get_global_state/cleanup_execution等接口。文件头部注释明确标注Stub — shared state removed in colony refactor当前实现中共享缓冲区仅作为兼容层存在get_recent_changes恒返回空列表真正的跨执行状态持久化由SessionStore的state.json与resume_session_id机制承载——这一演进在 execution_manager.py 的_run_execution中可见共享会话的执行会跳过初始/最终状态写入由主执行通过_write_progress()统一维护state.json。核心组件 2StreamRuntime 与执行追踪class StreamRuntime: def __init__(self, stream_id, storage, outcome_aggregator): # 按 execution_id 追踪运行而非单一 _current_run self._runs: dict[str, Run] {}现在多个执行可以并发运行而不冲突# 执行 A runtime.start_run(execution_idexec-A, goal_idsupport) runtime.decide(execution_idexec-A, intentclassify ticket, ...) # 执行 B并发无冲突 runtime.start_run(execution_idexec-B, goal_idsupport) runtime.decide(execution_idexec-B, intentclassify ticket, ...)实现现状仓库佐证仓库中的对应实现为 core/framework/host/stream_runtime.py 的StreamDecisionTracker——文档描述其关键差异正是通过execution_id并发追踪多个 run以及使用ConcurrentStorage保证线程安全的持久化。而每个入口点的执行编排由 core/framework/host/execution_manager.py 的ExecutionManager承担它持有_active_executionsExecutionContext字典、_execution_tasksasyncio.Task、_execution_results带 TTL 与数量上限的OrderedDict等簿记结构并内置复活机制resurrection——非致命错误下从失败节点自动重启默认最多max_resurrections3次_FATAL_ERROR_PATTERNS中列出的凭据/鉴权/配置类错误则直接判定致命、不复活。核心组件 3OutcomeAggregator 与跨流目标class OutcomeAggregator: def record_decision(self, stream_id, execution_id, decision) - None def record_outcome(self, stream_id, execution_id, decision_id, outcome) - None async def evaluate_goal_progress(self) - dict框架现在追踪所有流上的全部决策从而支持统一的目标进度评估跨执行的约束违规检测带正确归因的成功标准追踪。实现现状仓库佐证core/framework/host/outcome_aggregator.py 目前为兼容桩evaluate_goal_progress()返回占位结果AgentHost.get_goal_progress()见 agent_host.py会调用该聚合器完成跨流目标评估聚合器内部还接入EventBus以便发布GOAL_PROGRESS/GOAL_ACHIEVED/CONSTRAINT_VIOLATION等事件。核心组件 4EventBus 事件协调# 流 A 发布 await bus.publish(AgentEvent( typeEventType.EXECUTION_COMPLETED, stream_idwebhook, execution_idexec-123, data{ticket_resolved: True}, )) # 流 B 订阅 bus.subscribe( event_types[EventType.EXECUTION_COMPLETED], handleron_ticket_resolved, filter_streamwebhook, )流之间无需紧耦合或共享可变状态即可协调。实现现状仓库佐证core/framework/host/event_bus.py 定义了EventTypeStrEnum枚举覆盖执行生命周期EXECUTION_STARTED/EXECUTION_COMPLETED/EXECUTION_FAILED/EXECUTION_PAUSED/EXECUTION_RESUMED、状态变更STATE_CHANGED/STATE_CONFLICT、目标追踪GOAL_PROGRESS/GOAL_ACHIEVED/CONSTRAINT_VIOLATION、流生命周期、LLM 流式可观测性、工具调用、外部触发WEBHOOK_RECEIVED以及 Worker 生命周期WORKER_COMPLETED/WORKER_FAILED等 60 种事件类型。subscribe()支持filter_stream/filter_node/filter_graph三类过滤器AgentHost还提供subscribe_to_events()/unsubscribe_from_events()包装方法agent_host.py。另外execution_manager.py 中的GraphScopedEventBus会在每次publish()时自动给事件打上graph_id戳并代理订阅、历史与统计接口。四、什么时候该用 Tool决策对照表工具仍然是以下场景的正确选择外部系统集成——调用 API、数据库、服务副作用——发送邮件、创建资源数据检索——获取决策所需的信息。关键区分在于职责边界使用场景正确做法在多次执行之间协调SharedBufferManager追踪决策结果StreamRuntime OutcomeAggregator调用外部 APITool持久化业务数据Tool写入外部存储执行期间共享草稿状态StreamBuffer向其他流发布事件EventBus五、实战配置在 AgentHost 中注册多入口点当前仓库中AgentHost即文档所称AgentRuntime的现行实现通过EntryPointSpec注册每个入口点。以下综合 agent_host.py 的AgentRuntimeConfig与 execution_manager.py 的EntryPointSpec给出完整可运行示例runtime AgentHost( graphsupport_agent_graph, goalsupport_agent_goal, storage_pathPath(./storage), llmllm_provider, ) # 注册 Webhook 入口 runtime.register_entry_point(EntryPointSpec( idwebhook, nameZendesk Webhook, entry_nodeprocess-webhook, trigger_typewebhook, isolation_levelshared, )) # 注册 API 入口 runtime.register_entry_point(EntryPointSpec( idapi, nameAPI Handler, entry_nodeprocess-request, trigger_typeapi, isolation_levelshared, )) # 注册定时器入口每 5 分钟一次升级检查 runtime.register_entry_point(EntryPointSpec( idtimer, nameEscalation Check, entry_nodecheck-escalations, trigger_typetimer, trigger_config{interval_minutes: 5}, isolation_levelisolated, )) # 注册事件驱动入口监听其他流的完成事件 runtime.register_entry_point(EntryPointSpec( idevents, nameDelegate Responder, entry_nodehandle-delegated-task, trigger_typeevent, trigger_config{event_types: [execution_completed]}, isolation_levelshared, )) # 启动运行时会为每个入口点创建 ExecutionManager 流 await runtime.start() # 非阻塞触发立即返回 execution_id exec_1 await runtime.trigger(webhook, {ticket_id: 123}) exec_2 await runtime.trigger(api, {query: help}) # 检查目标进度 progress await runtime.get_goal_progress() print(fProgress: {progress[overall_progress]:.1%}) # 停止运行时 await runtime.stop()EntryPointSpec 字段详解字段类型默认值说明idstr必填入口点唯一标识同时作为流 IDnamestr必填入口点显示名entry_nodestr必填启动执行的图节点 ID必须存在于图中否则register_entry_point抛ValueErrortrigger_typestr必填webhook/api/timer/event/manualtrigger_configdict{}触发器附加配置见下文isolation_levelstrsharedisolated/shared/synchronized对应IsolationLevel枚举priorityint0优先级max_concurrentint10该入口点允许的最大并发执行数对应ExecutionManager内部的asyncio.Semaphoremax_resurrectionsint3非致命失败时的自动重启次数0 表示禁用AgentRuntimeConfig 关键参数agent_host.py 中的AgentRuntimeConfig提供如下可调项参数默认值说明max_concurrent_executions100全局最大并发执行数cache_ttl/batch_interval60.0/0.1传给ConcurrentStorage的缓存 TTL 与批量写入间隔max_history1000事件总线保留的历史事件数execution_result_max1000每个流保留的执行结果上限超出按 FIFO 淘汰execution_result_ttl_secondsNone执行结果保留 TTLNone 表示不过期idempotency_ttl_seconds/idempotency_max_keys300.0/10000trigger()幂等去重窗口与最大键数webhook_host/webhook_port127.0.0.1/8080Webhook 服务器监听地址与端口webhook_routes[]Webhook 路由列表非空才会启动服务器各触发器类型的 trigger_config 配置从AgentHost.start()与_start_timers()的实现agent_host.py可以归纳timer固定间隔模式{interval_minutes: 5, run_immediately: false, idle_timeout_seconds: 300}timerCron 表达式模式{cron: */5 * * * *, run_immediately: false, idle_timeout_seconds: 300}需要安装croniter缺失时启动抛RuntimeError并提示uv pip install croniterevent{event_types: [execution_completed, execution_failed], filter_stream: webhook, filter_node: ..., filter_graph: ..., exclude_own_graph: false}webhook对应AgentRuntimeConfig.webhook_routes每个路由形如{source_id: str, path: str, methods: [POST], secret: str|None}由 core/framework/host/webhook_server.py 的WebhookServer承载收到请求后会发布WEBHOOK_RECEIVED事件。仓库中自带一个真实的 timer 触发器示例见 examples/templates/email_inbox_management/triggers.json[ { id: email-timer, name: Scheduled Inbox Check, trigger_type: timer, trigger_config: { interval_minutes: 5 }, task: Fetch and process inbox emails according to the users rules } ]定时器的避让门控逻辑值得注意的一个工程细节AgentHost的定时器循环并非机械地到点触发而是带有避让门控——若任何流的执行正在活跃工作LLM/工具调用未超过idle_timeout_seconds默认 300 秒则跳过本次 tick避免定时任务打断正在进行的会话pause_timers()/resume_timers()可显式暂停/恢复所有定时器。这套逻辑在 agent_host.py 的_start_timers中实现并且在trigger()时会先经过 Pipeline 中间件限流、校验、成本防护等被拒时抛PipelineRejectedError。幂等触发与等待结果针对 Webhook 提供方超时重试的场景trigger()支持幂等键去重agent_host.py# 第一次触发缓存 execution_id exec_id await runtime.trigger(webhook, {ticket_id: 123}, idempotency_keywh-2026-0001) # 重试到达TTL 窗口内返回相同 execution_id不重复执行 exec_id_again await runtime.trigger(webhook, {ticket_id: 123}, idempotency_keywh-2026-0001) assert exec_id exec_id_again需要同步等待完成时使用trigger_and_wait(entry_point_id, input_data, timeout...)返回ExecutionResult或超时返回Noneagent_host.py。会话共享异步入口如何加入主会话AgentHost._get_primary_session_state()agent_host.py实现了关键的会话协作当 Webhook/定时器等异步入口以isolation_levelshared触发时会查找主会话如用户前端会话的活跃执行返回{resume_session_id: exec_id, data_buffer: {...}}从而复用同一个会话目录——记忆如用户定义的规则共享、日志落在同一会话目录且data_buffer只注入异步入口节点声明为输入的键避免上一轮过期的输出泄漏进来。而isolation_levelisolated的入口则使用独立会话目录并持久复用同一会话保证conversation_modecontinuous生效。多图扩展AgentHost还支持在运行中动态加载次级图add_graph()agent_host.py每个次级图获得独立的SessionStore与RuntimeLogStore会话/日志写入graphs/{graph_id}子目录共享同一个EventBus、state.json与数据目录并同样支持事件订阅与定时器入口。list_graphs()/remove_graph()/active_graph_id提供管理能力——这可以看作是多入口点在多 Agent图维度上的自然延伸。六、迁移指南从反模式到正确架构Before反模式# tools.py - 状态隐藏在工具里 tool def get_processing_count() - int: return redis.get(processing_count) or 0 tool def increment_processing_count() - int: return redis.incr(processing_count)After正确架构# 在节点执行中 async def execute(self, context, memory): # 从受管状态读取 count await memory.read(processing_count) or 0 # 带显式隔离级别更新 await memory.write( processing_count, count 1, scopeStateScope.STREAM, # 显式作用域 )迁移检查清单盘点现状找出所有在工具内部读写全局状态的代码Redis 计数器、内存字典、环境变量写等分类职责业务数据持久化保留在工具写入外部存储执行协调状态迁移到受管状态memory.read/writeStateScope流间通知改为EventBus发布/订阅设定隔离级别每个入口点在EntryPointSpec.isolation_level中显式声明isolated/shared/synchronized验证目标评估迁移后确认get_goal_progress()能正确归因跨流决策STATE_CONFLICT/CONSTRAINT_VIOLATION事件可按预期触发补充并发测试为每个入口点编写并发触发测试确认互不干扰、执行结果可独立追踪。七、总结两种方案对比方面Tools-as-State显式状态管理并发竞态条件隔离级别事务无执行级作用域可见性隐藏可观测测试需要 Mock设计上天然隔离目标追踪被破坏完整归因调试不透明可追踪多入口点架构的意义不止于让并发执行成为可能——它为可靠、可观测、目标驱动的 Agent 提供了地基让 Agent 能在生产环境中安全运行。在 Hive 当前代码库中这套设计的落地载体是 core/framework/host/agent_host.py生命周期与入口点管理、core/framework/host/execution_manager.py执行流与并发控制、core/framework/host/stream_runtime.py按执行 ID 的决策追踪、core/framework/host/event_bus.py事件协调以及 core/framework/host/webhook_server.py外部 HTTP 入口。相关行为测试可参考 core/tests/test_trigger_fires_into_queen.py、core/tests/test_node_conversation.py 与 core/tests/test_progress_db.py。需要提醒的是文档中所引的core/framework/runtime/目录路径在当前仓库中已不存在以上实现均位于core/framework/host/下阅读源码时请以实际路径为准。赞分享人工智能AI Agent多智能体MCP 服务工具调用浏览器控制【免费下载链接】hiveMulti-Agent Harness for Production AI项目地址https://gitcode.com/gh_mirrors/hive48/hive点击查看免费下载相关推荐Hive Agent Framework 声明式 Agent 构建指南agent.json 架构、节点/边配置与执行图实战Hive Agent Framework 声明式 Agent 构建指南agent.json 架构、节点/边配置与执行图实战 本篇技术指南以 Hive Agen人工智能AI Agent多智能体MCP 服务工具调用浏览器控制从单点到多活Apollo无状态服务架构的99.99%可用性实践从单点到多活Apollo无状态服务架构的99.99%可用性实践 你是否曾因配置中心单点故障导致全网服务不可用是否在机房断电时眼睁睁看着配置更新全部阻塞Ap配置中心后端微服务Agent OS 与 AgentMesh 统一治理架构从单机内核到分布式 Agent 生态的端到端治理实战Agent OS 与 AgentMesh 统一治理架构从单机内核到分布式 Agent 生态的端到端治理实战 Agent OS 与 AgentMesh 是 Ag人工智能AI AgentAI 安全治理策略引擎Agent 沙箱认证鉴权创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考