""" 工作流节点上下文查询工具（Python 端实现）。 3 个工具供 Hermes Agent 在执行工作流节点任务时调用，通过 HTTP 反向查询 Java 端的 `/internal/context/{path,output,meta}` 接口，获取前驱节点的路径图 / 输出 / 中间数据。 设计要点： - 工具永久注册到全局 registry（启动时一次性，幂等） - check_fn 检查 contextvar `_CONTEXT_QUERY_CTX`： - 已设置（含 context_query 的请求）→ 工具对 agent 可见 - 未设置（普通对话请求）→ 工具不可见，避免污染 - handler 内通过 contextvar 读取 base_url/token/run_id/visit_uuid， 避免在工具签名中暴露这些敏感字段给 LLM - 返回格式（用户决策 Q3）：自然语言摘要 + JSON 代码块 协议契约：参见 `docs/specs/internal-context-query-spec.md` 日志要点： - 入口：每个 handler 被调用时打印工具名 + 关键参数（不含 token） - HTTP：_http_post 打印 URL + 状态码 + 耗时 + 响应字节数 - 出口：返回内容的长度与摘要预览（前 100 字符） - token 永远不打印；visit_uuid 仅打印前 8 位 """ import json import logging import time import urllib.error import urllib.request from typing import Optional logger = logging.getLogger("internal-context-tools") # Java 端点路径（前缀 /internal/context） _PATH_ENDPOINT = "/internal/context/path" _OUTPUT_ENDPOINT = "/internal/context/output" _META_ENDPOINT = "/internal/context/meta" # HTTP 超时（秒）：内存查询亚秒级，DB 回退 < 300ms，5s 兜底足够 _HTTP_TIMEOUT = 5.0 # 工具所属 toolset，独立于 hermes-agent 内置 toolset，避免命名冲突 _TOOLSET = "workflow_context" # 日志预览截断长度（字符） _LOG_PREVIEW_LEN = 100 def _short_visit(uuid_str) -> str: """visit_uuid 截断为前 8 位用于日志（避免完整 UUID 噪声）。""" if not uuid_str: return "" return str(uuid_str)[:8] def _get_cfg(): """从 hermes_bridge 模块的 contextvar 读取当前请求的 context_query 配置。 返回 None 表示未启用（普通对话场景），handler 与 check_fn 都依赖此函数。 延迟 import 避免循环依赖（hermes_bridge 启动时会 import 本模块注册工具）。 """ try: from hermes_bridge import _CONTEXT_QUERY_CTX return _CONTEXT_QUERY_CTX.get(None) except Exception: return None def _http_post(base_url: str, token: str, path: str, payload: dict) -> tuple: """调用 Java 内部接口。 返回 (status_code, data_or_error)： - 成功：(200, ) - 业务失败：(non-200, {"message": "...", "code": ...}) - 网络/异常：(-1, {"message": "..."}) """ 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, }, ) start = time.time() try: with urllib.request.urlopen(req, timeout=_HTTP_TIMEOUT) as resp: body = resp.read().decode("utf-8") elapsed_ms = int((time.time() - start) * 1000) # Java Result: {code: 200, message: ..., data: T} parsed = json.loads(body) if body else {} code = parsed.get("code", resp.status) if code == 200: logger.info( "[CtxQuery] POST %s -> 200, %d bytes, %dms, payload=%s", url, len(body), elapsed_ms, _truncate_payload_for_log(payload), ) return 200, parsed.get("data") or {} logger.warning( "[CtxQuery] POST %s -> code=%d, %d bytes, %dms, msg=%s, payload=%s", url, code, len(body), elapsed_ms, parsed.get("message", ""), _truncate_payload_for_log(payload), ) return code, {"message": parsed.get("message", "未知错误"), "code": code} except urllib.error.HTTPError as e: elapsed_ms = int((time.time() - start) * 1000) try: err_body = e.read().decode("utf-8") parsed = json.loads(err_body) if err_body else {} msg = parsed.get("message", err_body) except Exception: parsed = {} msg = str(e) logger.warning( "[CtxQuery] POST %s -> HTTPError %d, %dms, msg=%s, payload=%s", url, e.code, elapsed_ms, msg, _truncate_payload_for_log(payload), ) return e.code, {"message": msg, "code": e.code} except Exception as e: elapsed_ms = int((time.time() - start) * 1000) logger.warning( "[CtxQuery] POST %s -> EXC %dms, err=%s, payload=%s", url, elapsed_ms, e, _truncate_payload_for_log(payload), ) return -1, {"message": f"请求失败: {e}", "code": -1} def _truncate_payload_for_log(payload: dict) -> str: """日志中打印 payload 的安全摘要（不含 token，visit_uuid 截断）。""" if not payload: return "{}" safe = {} for k, v in payload.items(): if k in ("visitUuid", "visit_uuid"): safe[k] = _short_visit(v) else: safe[k] = v s = json.dumps(safe, ensure_ascii=False) return s if len(s) <= 200 else s[:200] + "..." def _format_success(summary: str, data: dict) -> str: """格式化为「自然语言摘要 + JSON 代码块」。""" if not data: return summary return "{summary}\n\n```json\n{json}\n```".format( summary=summary, json=json.dumps(data, ensure_ascii=False, indent=2), ) def _format_error(message: str) -> str: """错误场景只返回自然语言描述（不带 JSON），避免 LLM 误以为有数据可引用。""" return message def _log_handler_return(tool_name: str, result: str, extra: str = ""): """打印 handler 返回值的长度与预览（前 100 字符）。""" preview = result[:_LOG_PREVIEW_LEN].replace("\n", "\\n") if len(result) > _LOG_PREVIEW_LEN: preview += "..." logger.info( "[CtxQuery] %s 返回 %d 字符%s: %s", tool_name, len(result), f" ({extra})" if extra else "", preview, ) # --------------------------------------------------------------------------- # Tool handlers # --------------------------------------------------------------------------- def _handle_get_predecessor_paths(args, **kw): tool_name = "get_predecessor_paths" cfg = _get_cfg() if cfg is None: logger.info("[CtxQuery] %s 被调用但 contextvar 未启用（降级返回错误）", tool_name) return _format_error("上下文查询未启用：本次运行未注入 context_query 配置") logger.info( "[CtxQuery] %s 被调用: run_id=%s, visit_uuid=%s", tool_name, cfg.get("run_id"), _short_visit(cfg.get("visit_uuid")), ) status, data = _http_post( cfg["base_url"], cfg["token"], _PATH_ENDPOINT, {"runId": cfg["run_id"], "visitUuid": cfg["visit_uuid"]}, ) if status == 401: _log_handler_return(tool_name, "token_invalid", "401") return _format_error("token 无效或已过期，无法查询前驱路径") if status == 403: _log_handler_return(tool_name, "forbidden", "403") return _format_error("runId 与 token 不匹配，禁止查询") if status == 404: _log_handler_return(tool_name, "not_found", "404") return _format_error( "未找到当前 visit 的前驱路径（runId={}, visitUuid={}）".format( cfg["run_id"], _short_visit(cfg["visit_uuid"]), ) ) if status != 200: _log_handler_return(tool_name, "error:" + str(status), str(status)) return _format_error("查询前驱路径失败：{}".format(data.get("message", "未知错误"))) node_count = len(data.get("nodeIds") or []) path_count = len(data.get("paths") or []) backtrack = data.get("backtrackEdgeKeys") or [] summary = "当前节点的前驱访问路径图：共 {} 个节点、{} 条路径".format( node_count, path_count, ) if backtrack: summary += "（含 {} 条智能回边）".format(len(backtrack)) summary += "。" result = _format_success(summary, data) _log_handler_return(tool_name, result, "nodes={},paths={}".format(node_count, path_count)) return result def _handle_get_node_output(args, **kw): tool_name = "get_node_output" cfg = _get_cfg() if cfg is None: logger.info("[CtxQuery] %s 被调用但 contextvar 未启用（降级返回错误）", tool_name) return _format_error("上下文查询未启用：本次运行未注入 context_query 配置") node_id = (args.get("node_id") or "").strip() field = args.get("field") iter_seq = args.get("iter_seq") logger.info( "[CtxQuery] %s 被调用: node_id=%s, field=%s, iter_seq=%s, run_id=%s", tool_name, node_id, field, iter_seq, cfg.get("run_id"), ) if not node_id: _log_handler_return(tool_name, "missing_node_id", "缺参") return _format_error("查询失败：node_id 是必填参数") payload = {"runId": cfg["run_id"], "nodeId": node_id} if field: payload["field"] = field if iter_seq is not None: payload["iterSeq"] = int(iter_seq) status, data = _http_post( cfg["base_url"], cfg["token"], _OUTPUT_ENDPOINT, payload, ) if status == 401: _log_handler_return(tool_name, "token_invalid", "401") return _format_error("token 无效或已过期，无法查询节点输出") if status == 403: _log_handler_return(tool_name, "forbidden", "403") return _format_error("runId 与 token 不匹配，禁止查询") if status == 404: _log_handler_return(tool_name, "not_found:" + node_id, "404") return _format_error( "节点 {} 尚未被访问（runId={}）".format(node_id, cfg["run_id"]), ) if status != 200: _log_handler_return(tool_name, "error:" + str(status), str(status)) return _format_error( "查询节点 {} 输出失败：{}".format( node_id, data.get("message", "未知错误"), ), ) iter_str = data.get("iterSeq", 1) status_code = data.get("status", "未知") field_hint = "字段 {}".format(field) if field else "整包输出" summary = "节点 {}（第 {} 次访问）的 {}，状态={}".format( node_id, iter_str, field_hint, status_code, ) if data.get("truncated"): summary += "（结果已截断）" summary += "。" result = _format_success(summary, data) _log_handler_return(tool_name, result, "node={},status={}".format(node_id, status_code)) return result def _handle_get_node_meta(args, **kw): tool_name = "get_node_meta" cfg = _get_cfg() if cfg is None: logger.info("[CtxQuery] %s 被调用但 contextvar 未启用（降级返回错误）", tool_name) return _format_error("上下文查询未启用：本次运行未注入 context_query 配置") node_id = (args.get("node_id") or "").strip() visit_uuid = (args.get("visit_uuid") or "").strip() kind = (args.get("kind") or "ALL").strip().upper() iter_seq = args.get("iter_seq") logger.info( "[CtxQuery] %s 被调用: node_id=%s, visit_uuid=%s, kind=%s, iter_seq=%s, run_id=%s", tool_name, node_id or "", _short_visit(visit_uuid) if visit_uuid else "", kind, iter_seq, cfg.get("run_id"), ) if not node_id and not visit_uuid: _log_handler_return(tool_name, "missing_target", "缺参") return _format_error("查询失败：node_id 和 visit_uuid 至少需要提供一个") payload = {"runId": cfg["run_id"], "kind": kind} if node_id: payload["nodeId"] = node_id if visit_uuid: payload["visitUuid"] = visit_uuid if iter_seq is not None: payload["iterSeq"] = int(iter_seq) status, data = _http_post( cfg["base_url"], cfg["token"], _META_ENDPOINT, payload, ) if status == 401: _log_handler_return(tool_name, "token_invalid", "401") return _format_error("token 无效或已过期，无法查询中间数据") if status == 403: _log_handler_return(tool_name, "forbidden", "403") return _format_error("runId 与 token 不匹配，禁止查询") if status == 404: target = _short_visit(visit_uuid) if visit_uuid else node_id _log_handler_return(tool_name, "not_found:" + target, "404") return _format_error( "未找到 {} 的访问记录（runId={}）".format(target, cfg["run_id"]), ) if status != 200: _log_handler_return(tool_name, "error:" + str(status), str(status)) return _format_error( "查询中间数据失败：{}".format(data.get("message", "未知错误")), ) target_id = data.get("nodeId") or node_id or _short_visit(visit_uuid) iter_str = data.get("iterSeq", 1) summary = "节点 {}（第 {} 次访问）的 {} 数据".format(target_id, iter_str, kind) if data.get("truncated"): summary += "（结果已截断）" summary += "。" result = _format_success(summary, data) _log_handler_return(tool_name, result, "target={},kind={}".format(target_id, kind)) return result # --------------------------------------------------------------------------- # Tool schemas (OpenAI Function Calling 格式，与 Java systemPrompt 描述对齐) # --------------------------------------------------------------------------- _GET_PREDECESSOR_PATHS_SCHEMA = { "name": "get_predecessor_paths", "description": ( "查询当前节点的所有前驱访问路径（含 visit 链、回边标识）。" "无需参数，自动使用当前请求的工作流运行上下文。" "适用场景：需要了解「当前节点运行前发生了什么」时调用。" ), "parameters": { "type": "object", "properties": {}, "required": [], }, } _GET_NODE_OUTPUT_SCHEMA = { "name": "get_node_output", "description": ( "查询某个前驱节点的输出（envelope.data 整包或单字段）。" "适用场景：需要精确引用某个前驱节点的输出变量时调用。" ), "parameters": { "type": "object", "properties": { "node_id": { "type": "string", "description": "目标节点 ID（必填）", }, "field": { "type": "string", "description": "可选；指定后只返回 envelope.data 内该字段的值", }, "iter_seq": { "type": "integer", "description": "可选；同节点多次访问时指定第几次（1-based），不填取最后一次", }, }, "required": ["node_id"], }, } _GET_NODE_META_SCHEMA = { "name": "get_node_meta", "description": ( "查询某个前驱节点的中间数据（思考过程 / 工具调用 / 待办清单 / 全部）。" "适用场景：需要理解前驱节点的思考过程或工具调用结果时调用。" ), "parameters": { "type": "object", "properties": { "node_id": { "type": "string", "description": "目标节点 ID（与 visit_uuid 二选一）", }, "visit_uuid": { "type": "string", "description": "指定具体访问的 UUID（与 node_id 二选一，优先使用）", }, "kind": { "type": "string", "enum": ["THINKING", "TOOL_CALLS", "TODO", "ALL"], "description": "返回内容类型，默认 ALL", }, "iter_seq": { "type": "integer", "description": "可选；同节点多次访问时指定第几次，不填取最后一次", }, }, "required": [], }, } # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def _check_context_active(*args, **kwargs) -> bool: """check_fn：仅当本次请求注入了 context_query 时工具对 agent 可见。""" return _get_cfg() is not None def register_internal_context_tools(): """一次性注册 3 个工具到全局 registry。 幂等：重复调用不会报错（相同 name + toolset 会覆盖）。 在 hermes_bridge 模块加载时调用一次即可。 """ try: from tools.registry import registry except ImportError: logger.warning("无法 import tools.registry，跳过工作流上下文工具注册") return try: registry.register( name="get_predecessor_paths", toolset=_TOOLSET, schema=_GET_PREDECESSOR_PATHS_SCHEMA, handler=_handle_get_predecessor_paths, check_fn=_check_context_active, emoji="🧭", ) registry.register( name="get_node_output", toolset=_TOOLSET, schema=_GET_NODE_OUTPUT_SCHEMA, handler=_handle_get_node_output, check_fn=_check_context_active, emoji="📤", ) registry.register( name="get_node_meta", toolset=_TOOLSET, schema=_GET_NODE_META_SCHEMA, handler=_handle_get_node_meta, check_fn=_check_context_active, emoji="🔍", ) logger.info("已注册工作流上下文查询工具（toolset=%s）", _TOOLSET) except Exception as e: logger.warning("注册工作流上下文工具失败: %s", e)