智能体运行时

本页描述 AgenticX 中的 `AgentRuntime` 执行路径（`agenticx/runtime/agent_runtime.py`）：think-act 循环、事件流、超时、确认门控、压缩、循环检测与钩子接入点。它与 `agenticx/core/agent_executor.py` 中较旧的 `AgentExecutor` 流水线不同——后者使用独立事件日志与上下文编译器。

智能体定义（`Agent` 模型）

主智能体记录为 agenticx.core.agent.Agent（Pydantic BaseModel）。常用于人设与租户的字段：

模型还携带 tools、LLM hooks、`max_iterations`（默认 25）、`max_retry_limit` 等配置。`Agent` 上 没有 `verbose` 布尔字段，也 没有 名为 `max_iter` 的字段——运行时工具循环 的迭代上限需单独配置。

运行智能体（`AgentExecutor`）

`agenticx.core.agent_executor.AgentExecutor` 是 `Agent` + `Task` 的控制流引擎：构建提示、调用配置的 `BaseLLMProvider`、解析动作、运行 `BaseTool` 实例（含并行批次）、应用 GuideRails，并可通过内部事件日志上的 `ContextCompiler.maybe_compact` 压缩上下文。

典型用法：

1from agenticx.core.agent import Agent
2from agenticx.core.agent_executor import AgentExecutor
3from agenticx.core.task import Task
4from agenticx.llms import OpenAIProvider
5
6agent = Agent(
7    name="Research Assistant",
8    role="Analyst",
9    goal="Summarize findings",
10    organization_id="my-org",
11)
12task = Task(
13    id="t1",
14    description="Research topic X",
15    expected_output="Short report",
16)
17executor = AgentExecutor(llm_provider=OpenAIProvider(model="gpt-4o"))
18result = executor.run(agent=agent, task=task)

`AgentExecutor.__init__` 接受 `max_iterations`（默认 50），与 Agent.max_iterations 独立。

!!! note "两套栈"

`AgentRuntime` 驱动 Studio/Desktop 聊天路径（OpenAI 风格消息、`STUDIO_TOOLS`、适配 SSE 的 `RuntimeEvent`）。`AgentExecutor` 驱动经典 task/agent 流水线。两者均有效；下文聚焦 `AgentRuntime`。

Think-act 循环（`AgentRuntime.run_turn`）

每个用户轮次运行 `AgentRuntime.run_turn`，最多循环 `max_tool_rounds` 次。一轮包括：

1. 消息清洗 — _sanitize_context_messages 修复历史，使每个带 tool_calls 的 assistant 消息后紧跟匹配的 tool 行（按 tool_call_id）。孤立 tool 行与悬空 tool_calls 会被丢弃，避免 provider 因序列断裂返回 HTTP 400。
2. 上下文压缩 — `ContextCompactor.maybe_compact` 可能用单一 `system` 摘要消息替换较早历史。成功时发出 `COMPACTION` 事件并运行 `on_compaction` 钩子。
3. LLM 调用 — 若 `llm.stream_with_tools` 存在且成功，运行时在 worker 线程中流式输出 content 与 tool_call_delta chunk，强制执行 first-feedback、invoke、heartbeat、hard 超时，规范化无效工具名（如 `"None"`）并修复流式 JSON 参数。任何失败时 回退 到带 `tools`、`tool_choice="auto"` 的 `llm.invoke`。`run_before_model` / `run_after_model` 钩子包裹调用。
4. 工具调用过滤 — 解析的 `tool_calls` 丢弃空名与 `name.lower() == "none"`。若模型未返回原生 tool call，运行时可从纯文本提取 inline tool call（当匹配允许工具集时）。
5. 工具分发 — 每次调用：`run_before_tool_call` 可阻塞。未知工具与无效名产生 `TOOL_RESULT`（常伴随 `ERROR`）但不执行。`_meta_only_names` 中的工具走 `dispatch_meta_tool_async`（需 `team_manager`）；其余通过运行时 `confirm_gate` 使用 `dispatch_tool_async`。分发器嵌套工具事件会 re-yield 为 `RuntimeEvent`。
6. 循环检测 — 每次工具执行后运行 `LoopDetector.record_call` / `check`。Warning 级别向 `messages` 注入 user 角色提醒；critical 级别发出 `ERROR` 并结束本轮。

若模型返回文本且无工具，运行时按需流式输出最终文本，运行 `run_on_agent_end`，yield `FINAL` 并返回。若 `max_tool_rounds` 耗尽仍无最终答案，meta 智能体得到 `ERROR`；非 meta 智能体得到带 checkpoint 元数据的 `SUBAGENT_PAUSED`。

运行时事件流（`EventType`）

agenticx/runtime/events.py 定义 Studio 与测试消费的字符串枚举：

超时配置

在 `agent_runtime.py` 中通过环境变量解析。单位为 秒。

传入 provider 的 LLM 请求超时 derived 为 `max(invoke, heartbeat, hard) + 15` 秒。

确认门控

Run Everything（Desktop 权限模式 `auto`）表示产品将会话内高风险操作视为预批准：确认应无重复弹窗地 resolve，与自动批准策略一致。

上下文压缩

`ContextCompactor.maybe_compact`（agenticx/runtime/compactor.py）：

• 当历史长度 > `threshold_messages`（默认 20，最小 8）或 总字符数 > `threshold_chars`（默认 48,000）时触发。
• 保留最后 `retain_recent_messages` 条消息（默认 8，最小 4）。
• 通过中文提示上的 `llm.invoke` 摘要前缀；失败时回退为截断片段。
• 返回以一条带 `[compacted] ...` 标签的 `system` 消息开头、外加保留尾部消息的列表。

循环检测

`LoopDetector`（agenticx/runtime/loop_detector.py）跟踪近期 (tool_name, args_signature) 对与进度标记。

`AgentRuntime` 以 `warning_threshold=4` 与 `critical_threshold=8`（构造参数）实例化。检测器包括：

• generic_repeat — 相同工具 + 相同参数重复。
• ping_pong — 尾部交替模式。
• no_progress — 多次调用无 artifact/scratchpad 进度等启发式。

Warning 添加提醒消息；critical 以 `ERROR` 停止运行。

运行时配置

事件钩子（`HookRegistry` / `AgentHook`）

已注册钩子（agenticx/runtime/hooks/__init__.py）按优先级顺序运行：

`AgentRuntime` 在可用时以低优先级注册 `MemoryHook`。

相关

• 编排 — 多步骤工作流与协调。
• 工具 — 工具定义与分发。
• 记忆 — 长程回忆与钩子。