agent-management/docs/plans/node-context-runtime-tools-plan.md

节点上下文运行时工具实施方案（Bridge 侧）

状态：已完成 关联：

• node-context-query-tool-plan.md（顶层设计：方案 C，Java 端工具网关）
• ../specs/internal-context-query-spec.md（HTTP 接口契约） 创建日期：2026-08-13

一、背景

node-context-query-tool-plan.md 选定方案 C（Java 端工具网关）：LLM 调用工具 → Python Bridge HTTP 回调 Java 内部接口 → Java 服务查询内存/DB → 返回视图。

Java 侧已 100% 就绪：

能力	文件	状态
路径图查询	engine/NodeContextQueryService.queryPath	已实现
输出查询	engine/NodeContextQueryService.queryOutput	已实现
中间数据查询	engine/NodeContextQueryService.queryMeta	已实现
HTTP 端点	controller/InternalContextQueryController	已实现（3 个 POST）
Token 鉴权	engine/InternalContextTokenService	已实现
配置开关	config/HermesContextQueryProperties	已实现
Agent 执行器注入	engine/executor/HermesAgentExecutor	已调用 support.buildContextQuery(...)
SmartAction 执行器注入	engine/executor/HermesSmartActionExecutor	已调用 support.buildContextQuery(...)
Bridge 协议字段	engine/hermes/HermesBridgeClient	已序列化 context_query 到 body
systemPrompt 引导	HermesBridgeClient.buildContextToolsSection	已注入 3 个工具说明

Python 侧缺失：backend/hermes-bridge/hermes_bridge.py 未实现任何 workflow_context 相关工具——Java 注入的 context_query 字段被忽略，LLM 看到工具说明但实际无工具可调。

二、目标

在 hermes_bridge.py 中实现并注册 3 个工具，名称与 Java systemPrompt 中已声明的完全一致：

工具名	参数	对应 Java 端点
get_predecessor_paths	（无）	POST /internal/context/path
get_node_output	node_id（必填）、field?、iter_seq?	POST /internal/context/output
get_node_meta	node_id 或 visit_uuid、kind?、iter_seq?	POST /internal/context/meta

返回格式（用户决策 Q3）：自然语言摘要 + JSON，让 LLM 既易于理解又便于精确引用。

注册条件：仅当请求 body 含 context_query 字段时注册，避免无工作流上下文的纯 Agent 对话场景被污染。

三、协议契约回顾

3.1 Java → Python（/run body 新增字段）

{
  "user_message": "...",
  "system_prompt": "...",
  "context_query": {
    "base_url": "http://127.0.0.1:8080",
    "token": "一次性鉴权 token",
    "run_id": "abc12345",
    "visit_uuid": "uuid-of-current-visit"
  }
}

3.2 Python → Java（工具内部 HTTP 回调）

3 个端点，统一 Header：X-Context-Token: <token>，统一 Body 字段：runId + visitUuid（path 必填 visitUuid；output/meta 二选一）。

详见 ../specs/internal-context-query-spec.md。

3.3 返回格式（自然语言摘要 + JSON）

每个工具返回字符串，结构：

[摘要句子]

json { ...原始 JSON... }

例如 get_node_output 返回：

节点 llm_1（第 1 次访问）的输出，状态=200：

json { "node_id": "llm_1", "iter_seq": 1, "visit_uuid": "abc-uuid", "status": 200, "message": "执行成功", "data": { "summary": "..." } }

错误场景（HTTP 非 200、token 失效、节点未访问）：返回简洁的自然语言错误描述（不带 JSON 体），如：

查询失败：节点 llm_1 未访问过（runId=abc12345）

四、实施阶段

阶段 1：扩展 RunRequest 与 contextvar

文件：backend/hermes-bridge/hermes_bridge.py

改动：

1. 新增 ContextQueryConfig Pydantic 子模型：

   class ContextQueryConfig(BaseModel):
      base_url: str
      token: str
      run_id: str
      visit_uuid: str
   

2. RunRequest 增加字段：

   context_query: Optional[ContextQueryConfig] = None
   

3. 新增模块级 contextvar：

   import contextvars
   _CONTEXT_QUERY_CTX: contextvars.ContextVar[Optional[ContextQueryConfig]] = \
      contextvars.ContextVar("context_query", default=None)
   

4. _run_agent_stream 接收 context_query 参数，在 _run 内部 _CONTEXT_QUERY_CTX.set(cq)（与 _SESSION_CWD.set(wd) 同位置、同样在 finally 中 reset）。

验证：单元测试直接调用 _run_agent_stream 时不报字段缺失；contextvar 在线程外不可见。

阶段 2：实现 workflow_context 工具模块

新文件：backend/hermes-bridge/workflow_context_tools.py

职责：3 个工具的纯实现，无注册逻辑（注册放在 hermes_bridge.py，便于按需注册）。

核心代码骨架：

"""Workflow context query tools for Hermes Bridge.

工具实现：通过 HTTP 回调 Java 内部接口查询前驱节点的运行信息。
注册由 hermes_bridge.py 在 _run_agent_stream 中按需完成（仅当请求含 context_query 时）。
"""
import json
import logging
import urllib.request
import urllib.error
from typing import Optional

logger = logging.getLogger("workflow-context-tools")

# Java 端点路径
_PATH_ENDPOINT = "/internal/context/path"
_OUTPUT_ENDPOINT = "/internal/context/output"
_META_ENDPOINT = "/internal/context/meta"

# HTTP 超时（秒）：内存查询亚秒级，DB 回退 < 300ms，给 5s 兜底
_HTTP_TIMEOUT = 5.0


def _http_post(base_url: str, token: str, path: str, payload: dict) -> tuple[int, dict]:
    """调用 Java 内部接口，返回 (status_code, response_json)。

    失败场景统一返回 (status, {"error": "..."})，由调用方包装为自然语言错误。
    """
    url = base_url.rstrip("/") + path
    data = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        url, data=data, method="POST",
        headers={"Content-Type": "application/json", "X-Context-Token": token},
    )
    try:
        with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp:
            body = resp.read().decode("utf-8")
            return resp.status, json.loads(body) if body else {}
    except urllib.error.HTTPError as e:
        try:
            err_body = e.read().decode("utf-8")
            err_msg = json.loads(err_body).get("message", err_body)
        except Exception:
            err_msg = str(e)
        return e.code, {"error": err_msg}
    except Exception as e:
        return -1, {"error": f"请求失败: {e}"}


def _format_result(summary: str, data: Optional[dict]) -> str:
    """格式化为「自然语言摘要 + JSON」格式。"""
    if data is None or "error" in data:
        return summary  # 错误场景只返回自然语言
    return f"{summary}\n\n```json\n{json.dumps(data, ensure_ascii=False, indent=2)}\n```"


def get_predecessor_paths() -> str:
    cfg = _resolve_cfg_or_raise()
    if isinstance(cfg, str):  # 错误提示
        return cfg
    status, resp = _http_post(
        cfg.base_url, cfg.token, _PATH_ENDPOINT,
        {"runId": cfg.run_id, "visitUuid": cfg.visit_uuid},
    )
    if status != 200 or "error" in resp:
        return f"查询路径图失败（HTTP {status}）: {resp.get('error', '未知错误')}"
    paths = resp.get("data") or {}
    node_count = len(paths.get("nodeIds") or [])
    path_count = len(paths.get("paths") or [])
    summary = f"当前节点前驱路径图：共 {node_count} 个节点，{path_count} 条路径。"
    return _format_result(summary, paths)


def get_node_output(node_id: str, field: Optional[str] = None,
                     iter_seq: Optional[int] = None) -> str:
    cfg = _resolve_cfg_or_raise()
    if isinstance(cfg, str):
        return cfg
    payload = {"runId": cfg.run_id, "nodeId": node_id}
    if field is not None:
        payload["field"] = field
    if iter_seq is not None:
        payload["iterSeq"] = iter_seq
    status, resp = _http_post(cfg.base_url, cfg.token, _OUTPUT_ENDPOINT, payload)
    if status == 404:
        return f"节点 {node_id} 未访问过（runId={cfg.run_id}）"
    if status != 200 or "error" in resp:
        return f"查询节点 {node_id} 输出失败（HTTP {status}）: {resp.get('error', '')}"
    data = resp.get("data") or {}
    iter_str = f"第 {data.get('iterSeq', 1)} 次访问" if data else "未知访问序"
    summary = f"节点 {node_id}（{iter_str}）的输出，状态={data.get('status', '未知')}。"
    return _format_result(summary, data)


def get_node_meta(node_id: Optional[str] = None, visit_uuid: Optional[str] = None,
                   kind: str = "ALL", iter_seq: Optional[int] = None) -> str:
    cfg = _resolve_cfg_or_raise()
    if isinstance(cfg, str):
        return cfg
    if not node_id and not visit_uuid:
        return "查询失败：node_id 和 visit_uuid 至少需要提供一个"
    payload = {"runId": cfg.run_id, "kind": kind.upper()}
    if node_id:
        payload["nodeId"] = node_id
    if visit_uuid:
        payload["visitUuid"] = visit_uuid
    if iter_seq is not None:
        payload["iterSeq"] = iter_seq
    status, resp = _http_post(cfg.base_url, cfg.token, _META_ENDPOINT, payload)
    if status == 404:
        target = visit_uuid or node_id
        return f"未找到 {target} 对应的访问记录（runId={cfg.run_id}）"
    if status != 200 or "error" in resp:
        return f"查询中间数据失败（HTTP {status}）: {resp.get('error', '')}"
    data = resp.get("data") or {}
    target_id = data.get("nodeId") or node_id or visit_uuid
    summary = f"节点 {target_id}（第 {data.get('iterSeq', 1)} 次访问）的 {kind} 数据。"
    return _format_result(summary, data)


def _resolve_cfg_or_raise():
    """从 contextvar 读取配置；缺失时返回错误字符串。"""
    from hermes_bridge import _CONTEXT_QUERY_CTX
    cfg = _CONTEXT_QUERY_CTX.get(None)
    if cfg is None:
        return "上下文查询未启用：本次运行未注入 context_query 配置"
    return cfg

关键设计点：

• 使用标准库 urllib，避免引入新依赖（FastAPI 已在但额外 httpx 不必要）
• 工具内通过 contextvar 读取 cfg，无需在工具签名中暴露 token/runId（避免 LLM 看到敏感信息）
• 错误场景不返回 JSON，避免 LLM 误以为有数据可引用
• 成功场景返回 Markdown 代码块包裹的 JSON，LLM 既看到摘要又能精确解析

验证：单元测试 mock urllib，验证 3 个工具的 HTTP 调用参数与返回格式。

阶段 3：动态工具注册

文件：backend/hermes-bridge/hermes_bridge.py

改动：

在 _run_agent_stream._run 内部（设置 _CONTEXT_QUERY_CTX 之后），动态把 3 个工具注入 agent：

# 工具注册的具体方式取决于 hermes-agent 的 API：
# 假设方式 A：agent.tools 是一个 list/dict，可直接 append
# 假设方式 B：调用 agent.register_tool(name, fn, schema)
# 实施前需查 hermes-agent 源码确认
registered_tools = []
if context_query is not None:
    registered_tools = _register_workflow_context_tools(agent)
try:
    # 原有 agent.run_conversation 调用
    ...
finally:
    _unregister_workflow_context_tools(agent, registered_tools)

待实施时确认：阅读 backend/hermes-agent/ 源码确定 hermes-agent 的工具注册 API。候选模式：

• 装饰器模式：@agent.tool 在 agent 创建后注册（不适合动态注册）
• 工厂注入：AIAgent(tools=[...]) 在构造时传入（要求每次按需重建 agent，与 LRU 池冲突）
• 运行时注册：agent.tool_registry.add(...) / agent.add_tool(...)（最理想）

若 hermes-agent 不支持运行时动态注册，则降级为：每次请求含 context_query 时不复用 LRU 池，新建临时 agent 并传入完整工具列表。

验证：integration test——含 context_query 的请求触发 3 个工具可用；不含 context_query 的请求不暴露这 3 个工具。

阶段 4：systemPrompt 引导校验

文件：无（复用现有）

Java 端 HermesBridgeClient.buildContextToolsSection 已在 systemPrompt 中注入工具说明，无需 Python 侧重复注入。本阶段仅验证：

• Java 注入的工具名（get_predecessor_paths / get_node_output / get_node_meta）与 Python 注册的工具名完全一致
• 参数名匹配（snake_case，与 Python 函数签名一致）

阶段 5：测试与文档

测试：

1. Python 侧单元测试 test_workflow_context_tools.py：

   • mock urllib.request.urlopen，验证 3 个工具的 HTTP 调用参数
   • 验证错误场景（404/500/网络异常）返回自然语言描述
   • 验证成功场景返回「摘要 + JSON」格式

2. 端到端验证（手动，由用户执行）：

   • 启动后端 + Bridge
   • 创建工作流：LLM 节点 A → LLM 节点 B（B 的 prompt 中要求调用 get_predecessor_paths 与 get_node_output）
   • 运行工作流，确认 B 节点能调用工具并拿到 A 的输出

文档：

• 本计划文件
• 在 ../specs/internal-context-query-spec.md 末尾追加 Python 工具实现说明（如需要）

五、设计决策记录

决策点	选项	决定	理由
返回格式（Q3）	纯 JSON / 自然语言摘要+JSON / 纯自然语言	摘要+JSON	既有可读性又便于 LLM 精确引用
SmartAction 覆盖（Q6）	仅 Agent / 全覆盖	全覆盖	Java 侧已实现；Python 侧工具按 context_query 自动注册，与执行器类型无关
审计日志（Q7）	现有 tool_start/tool_end / 新增类型	现有	工具调用已通过 on_tool_start/on_tool_end 自动落 ExecutionLog.TOOL_CALL/TOOL_RESULT
HTTP 客户端	httpx / urllib / requests	urllib	标准库零依赖；查询量低，性能不是瓶颈
工具名风格	snake_case / camelCase	snake_case	Python 惯例；与 hermes-agent 其他工具一致
contextvar vs 参数注入	工具签名暴露 / contextvar 隐藏	contextvar	避免向 LLM 暴露 token/runId，防止误用

六、风险与缓解

风险	级别	缓解
hermes-agent 不支持运行时动态工具注册	中	阶段 3 降级：含 context_query 时跳过 LRU 复用，临时构造 agent
工具调用失败导致 LLM 陷入重试循环	中	错误返回自然语言描述（非 JSON），LLM 能识别为终态；超时 5s 兜底
LLM 滥用查询接口消耗 token	低	Java systemPrompt 已声明「调用频率 3-5 次」；Java 侧 NodeContextQueryService 有 maxBytes 截断与单 run 50 次告警
token 在长任务中过期	低	Java 端 token TTL（默认 3600s）；HermesExecutorSupport 在节点 finally 中立即失效

七、估计复杂度

阶段	工时
1 RunRequest + contextvar	1h
2 工具模块实现	2-3h
3 动态注册（含 hermes-agent API 调研）	2-4h
4 systemPrompt 校验	0.5h
5 测试与端到端	2-3h
合计	7-11h

八、实施进度

阶段	状态	完成时间	备注
1	已完成	2026-08-13	RunRequest + context_query 字段；_CONTEXT_QUERY_CTX contextvar
2	已完成	2026-08-13	internal_context_tools.py 实现 3 个工具
3	已完成	2026-08-13	force_new=True 强制新建 agent + check_fn 按 contextvar 控制可见性
4	已完成	2026-08-13	复用 Java 端 systemPrompt（已注入工具说明）
5	部分完成	2026-08-13	dry-run 通过；端到端测试待用户在运行环境中手动验证

九、实施总结

9.1 实际落地的设计决策（相对原计划的调整）

决策点	原计划	实际	理由
工具注册时机	每次请求动态注册 + finally 注销	进程启动时一次性注册 + check_fn 控制可见性	registry.register 幂等；check_fn 已能按 contextvar 控制是否进入 agent.tools，无需反复注销
LRU 复用策略	未定	含 context_query 时强制新建（force_new=True）	已缓存 agent 的 agent.tools 是构造时快照，无法刷新；新建 agent 才能拉到工具 schema
contextvar set 位置	未定	在 _run 线程内部 set	contextvars 不跨线程自动传播，必须在工具实际执行的线程内 set/reset

9.2 文件清单

新增文件：

• backend/hermes-bridge/internal_context_tools.py：3 个工具的 handler、schema、check_fn、注册函数

修改文件：

• backend/hermes-bridge/hermes_bridge.py：
  • 顶部新增 import contextvars
  • 新增 _CONTEXT_QUERY_CTX contextvar 定义
  • 启动时调用 internal_context_tools.register_internal_context_tools()
  • RunRequest 新增 context_query: Optional[dict] 字段
  • _get_or_create_agent 新增 force_new 参数，force_new=True 时不查 LRU、不入 LRU
  • _run_agent_stream 新增 context_query 参数，在 _run 线程内 set/reset contextvar
  • /run 端点：req.context_query 非空时 force_new=True，传 context_query 给 _run_agent_stream

计划文档：

• docs/plans/node-context-runtime-tools-plan.md（本文件）

9.3 验证记录

• Python 语法校验：hermes_bridge.py + internal_context_tools.py 全部通过 ast.parse
• 跨语言字段名一致性：Java systemPrompt 工具名、HTTP 端点路径、context_query 子字段、请求 payload key 全部对齐
• dry-run 测试（mock tools.registry）：
  • 注册 3 个工具成功
  • check_fn 在无 contextvar 时返回 False（工具对 agent 不可见）
  • check_fn 在有 contextvar 时返回 True（工具进入 agent.tools）
  • _format_success 输出「自然语言摘要 + JSON 代码块」格式
  • _format_error 仅返回自然语言描述（无 JSON 体，避免 LLM 误引用）
  • handler 在无 contextvar 时优雅降级返回错误描述

9.4 端到端验证待办（用户手动）

1. 启动 backend + hermes-bridge
2. 在工作流编辑器中创建：节点 A（LLM，输出"hello"）→ 节点 B（智能体，prompt 中明确要求"调用 get_predecessor_paths 与 get_node_output 查询 A 的输出"）
3. 运行工作流，观察：
   • 节点 B 的执行日志中出现 get_predecessor_paths 与 get_node_output 工具调用
   • 工具返回「自然语言摘要 + JSON」格式
   • 节点 B 最终输出能正确引用 A 的结果
4. 验证对话页面（无 context_query）的 agent 不暴露这 3 个工具