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

> 状态：已完成
> 关联：
> - [node-context-query-tool-plan.md](./node-context-query-tool-plan.md)（顶层设计：方案 C，Java 端工具网关）
> - [../specs/internal-context-query-spec.md](../specs/internal-context-query-spec.md)（HTTP 接口契约）
> 创建日期：2026-08-13

## 一、背景

[node-context-query-tool-plan.md](./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 新增字段）

```json
{
  "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](../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 子模型：
   ```python
   class ContextQueryConfig(BaseModel):
       base_url: str
       token: str
       run_id: str
       visit_uuid: str
   ```

2. `RunRequest` 增加字段：
   ```python
   context_query: Optional[ContextQueryConfig] = None
   ```

3. 新增模块级 contextvar：
   ```python
   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`，便于按需注册）。

**核心代码骨架**：

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

```python
# 工具注册的具体方式取决于 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](../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 个工具