#!/usr/bin/env python3 """ Hermes Bridge - 将 Hermes AIAgent 封装为 HTTP API，供 Spring Boot 工作流引擎调用。 启动方式： python hermes_bridge.py --port 18731 python hermes_bridge.py --port 18731 --hermes-home /path/to/.hermes API： POST /run 运行一次 Agent 对话（SSE 流式响应） GET /health 健康检查 """ import argparse import json import logging import os import sys import threading import time import uuid from collections import OrderedDict from pathlib import Path from typing import Optional # --------------------------------------------------------------------------- # 强制 stdout/stderr 使用 UTF-8（Windows 默认 cp936 会导致中文输出乱码） # 必须在任何 print/logging 之前执行 # --------------------------------------------------------------------------- for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass # --------------------------------------------------------------------------- # 父进程守护：Java 后端被 kill 后，子进程自动退出，释放端口与资源 # --------------------------------------------------------------------------- def _start_parent_watcher(): """启动守护线程，当父进程退出时自杀。""" try: import psutil except ImportError: logging.warning("未安装 psutil，无法监听父进程状态；Java 退出后子进程可能残留") return try: parent = psutil.Process(os.getppid()) except Exception: return def _watch(): while True: time.sleep(2) try: if not parent.is_running() or parent.status() == psutil.STATUS_ZOMBIE: logging.info("父进程已退出，Hermes Bridge 自动终止") os._exit(0) except Exception: logging.info("父进程状态不可获取，Hermes Bridge 自动终止") os._exit(0) watcher = threading.Thread(target=_watch, daemon=True, name="parent-watcher") watcher.start() _start_parent_watcher() # --------------------------------------------------------------------------- # Hermes 项目路径注入（让 Python 能找到 hermes-agent 的源码） # --------------------------------------------------------------------------- BRIDGE_DIR = Path(__file__).resolve().parent HERMES_AGENT_DIR = BRIDGE_DIR.parent / "hermes-agent" if HERMES_AGENT_DIR.is_dir() and str(HERMES_AGENT_DIR) not in sys.path: sys.path.insert(0, str(HERMES_AGENT_DIR)) # --------------------------------------------------------------------------- # 日志 # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", stream=sys.stderr, ) logger = logging.getLogger("hermes-bridge") # --------------------------------------------------------------------------- # FastAPI app # --------------------------------------------------------------------------- try: from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel except ImportError: logger.error("请先安装 FastAPI: pip install fastapi uvicorn[standard]") sys.exit(1) app = FastAPI(title="Hermes Bridge", version="1.0.0") # --------------------------------------------------------------------------- # 共享密钥（X-Bridge-Token 头校验） # --------------------------------------------------------------------------- # 启动时从环境变量读取；为空则禁用认证（仅依赖 127.0.0.1 绑定） _BRIDGE_AUTH_TOKEN = os.environ.get("HERMES_BRIDGE_AUTH_TOKEN", "").strip() @app.middleware("http") async def _verify_token(request, call_next): """校验 X-Bridge-Token 头（/health 不校验）""" if _BRIDGE_AUTH_TOKEN and request.url.path != "/health": token = request.headers.get("X-Bridge-Token", "") if token != _BRIDGE_AUTH_TOKEN: from fastapi.responses import JSONResponse return JSONResponse(status_code=401, content={"detail": "invalid or missing X-Bridge-Token"}) return await call_next(request) # --------------------------------------------------------------------------- # 全局 Agent 池（按 session_id::hermes_home 复用，有界 LRU） # --------------------------------------------------------------------------- _agent_lock = threading.Lock() _agents: "OrderedDict[str, object]" = OrderedDict() _AGENT_CACHE_MAX = 8 # 最大缓存实例数；超出按 FIFO 淘汰，防止无界增长 # --------------------------------------------------------------------------- # 请求模型 # --------------------------------------------------------------------------- class RunRequest(BaseModel): """运行 Agent 对话的请求体""" system_prompt: Optional[str] = None # 系统提示（如 SKILL.md 内容） user_message: str # 用户消息 max_iterations: int = 100 # 最大工具调用轮次（兜底默认值，由后端配置覆盖） session_id: Optional[str] = None # 会话 ID（复用 Agent 实例） hermes_home: Optional[str] = None # 本次运行使用的 HERMES_HOME 路径（其下应有 skills 子目录） working_dir: Optional[str] = None # agent 工具调用的实际工作目录（终端、文件读写以它为根） model_id: Optional[str] = None # 节点指定的模型 ID base_url: Optional[str] = None # 节点指定的模型 base URL api_key: Optional[str] = None # 节点指定的模型 API Key model_name: Optional[str] = None # 节点指定的模型标识 # --------------------------------------------------------------------------- # Agent 工厂 # --------------------------------------------------------------------------- def _get_or_create_agent(session_id: str, max_iterations: int, hermes_home: Optional[str] = None, model_id: Optional[str] = None, base_url: Optional[str] = None, api_key: Optional[str] = None, model_name: Optional[str] = None): """获取或创建 AIAgent 实例 hermes_home 用于覆盖本次运行的 HERMES_HOME（其下应有 skills 子目录）， 实现按工作流运行目录隔离的 Skill 加载。 model_id/base_url/api_key/model_name 支持按节点动态切换模型； 未提供时回退到环境变量默认配置。 """ # 缓存键同时绑定 hermes_home 与模型 ID，避免跨运行目录/模型复用 agent cache_key = f"{session_id}::{hermes_home or ''}::{model_id or 'default'}" if session_id else None if cache_key: with _agent_lock: if cache_key in _agents: _agents.move_to_end(cache_key) # LRU 提升到最新 return _agents[cache_key] # 在导入/创建 AIAgent 之前显式覆盖 HERMES_HOME，使 get_hermes_home() 返回正确路径 if hermes_home: os.environ["HERMES_HOME"] = str(hermes_home) logger.info("使用请求指定的 HERMES_HOME: %s", hermes_home) # 安装沙箱补丁（幂等），强制 hermes-agent 文件操作限制在 _SESSION_CWD 内 try: import sandbox_patch sandbox_patch.install_sandbox_patches() except Exception as e: logger.warning("沙箱补丁加载异常（继续但不保证隔离）: %s", e) try: from run_agent import AIAgent except ImportError: raise ImportError( "无法导入 hermes-agent。请确保 hermes-agent 目录存在或已安装 hermes-agent 包。\n" f"尝试的路径: {HERMES_AGENT_DIR}" ) # 优先使用请求中传入的模型配置，其次回退到环境变量 effective_base_url = base_url or os.environ.get("HERMES_BRIDGE_BASE_URL", "") effective_api_key = api_key or os.environ.get("HERMES_BRIDGE_API_KEY", "") effective_model = model_name or os.environ.get("HERMES_BRIDGE_MODEL", "") # 尝试从 Hermes config.yaml 读取（如果未通过请求或环境变量指定） if not effective_base_url or not effective_api_key: _load_hermes_config() effective_base_url = effective_base_url or os.environ.get("HERMES_BRIDGE_BASE_URL", "") effective_api_key = effective_api_key or os.environ.get("HERMES_BRIDGE_API_KEY", "") effective_model = effective_model or os.environ.get("HERMES_BRIDGE_MODEL", "") if not effective_api_key: raise ValueError("未配置 API Key。请设置 HERMES_BRIDGE_API_KEY 环境变量或在 Hermes config.yaml 中配置。") logger.info( "创建 AIAgent: session=%s, model_id=%s, model=%s, base_url=%s", session_id, model_id or "默认", effective_model or "默认", effective_base_url or "默认" ) kwargs = dict( base_url=effective_base_url or None, api_key=effective_api_key, model=effective_model or None, max_iterations=max_iterations, quiet_mode=True, tool_progress_mode="off", skip_context_files=True, load_soul_identity=False, skip_memory=True, ) agent = AIAgent(**kwargs) if cache_key: with _agent_lock: _agents[cache_key] = agent # 超出上限按 FIFO 淘汰最旧实例 while len(_agents) > _AGENT_CACHE_MAX: _agents.popitem(last=False) return agent def _load_hermes_config(): """尝试从 Hermes 的 config.yaml 和 .env 加载配置""" try: hermes_home = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) # 加载 .env env_file = hermes_home / ".env" if env_file.exists(): for line in env_file.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") # 只设置 Bridge 相关的环境变量（如果尚未设置） if key in ("OPENAI_API_KEY", "OPENROUTER_API_KEY") and not os.environ.get("HERMES_BRIDGE_API_KEY"): os.environ["HERMES_BRIDGE_API_KEY"] = value if key == "OPENAI_BASE_URL" and not os.environ.get("HERMES_BRIDGE_BASE_URL"): os.environ["HERMES_BRIDGE_BASE_URL"] = value # 加载 config.yaml config_file = hermes_home / "config.yaml" if config_file.exists(): import yaml with open(config_file, "r", encoding="utf-8") as f: config = yaml.safe_load(f) or {} model_cfg = config.get("model") or {} provider = config.get("provider") or {} if not os.environ.get("HERMES_BRIDGE_MODEL") and model_cfg.get("name"): os.environ["HERMES_BRIDGE_MODEL"] = model_cfg["name"] if not os.environ.get("HERMES_BRIDGE_BASE_URL") and provider.get("base_url"): os.environ["HERMES_BRIDGE_BASE_URL"] = provider["base_url"] if not os.environ.get("HERMES_BRIDGE_API_KEY") and provider.get("api_key"): os.environ["HERMES_BRIDGE_API_KEY"] = provider["api_key"] except Exception as e: logger.warning("加载 Hermes 配置失败: %s", e) # --------------------------------------------------------------------------- # SSE 事件流 # --------------------------------------------------------------------------- def _run_agent_stream(agent, system_prompt: str, user_message: str, run_id: str, working_dir: Optional[str] = None, session_id: Optional[str] = None): """运行 Agent 并生成 SSE 事件流""" queue = [] done_event = threading.Event() def on_stream(delta: str): """接收流式文本片段""" queue.append(("text", {"content": delta})) def on_tool_start(tool_name: str, args: dict): """工具开始""" queue.append(("tool_start", {"tool": tool_name, "args": args})) def on_tool_end(tool_name: str, result: str): """工具结束""" summary = result[:500] if result else "" queue.append(("tool_end", {"tool": tool_name, "result_summary": summary})) def _run(): # 通过 contextvar 设置本次运行的 agent 工作目录； # hermes-agent 内 resolve_agent_cwd() 会优先读取此值， # 影响 agent 工具（终端、文件读写）的当前目录。 cwd_token = None original_cwd = None effective_task_id = session_id or "default" if working_dir: wd = str(working_dir) # 同时切换进程 cwd，覆盖那些直接调用 os.getcwd() 的工具 try: original_cwd = os.getcwd() os.chdir(wd) logger.info("[%s] 切换进程 cwd: %s", run_id, wd) except Exception as e: logger.warning("[%s] 切换 cwd 失败: %s", run_id, e) # 设置 TERMINAL_CWD 作为 local environment 的 fallback cwd os.environ["TERMINAL_CWD"] = wd # 注册 task env override，强制更新（或创建）该 session 的 terminal environment cwd try: from tools.terminal_tool import register_task_env_overrides register_task_env_overrides(effective_task_id, {"cwd": wd}) logger.info("[%s] 注册 terminal cwd override for task %s: %s", run_id, effective_task_id, wd) except Exception as e: logger.warning("[%s] 注册 terminal cwd override 失败: %s", run_id, e) try: from agent.runtime_cwd import _SESSION_CWD cwd_token = _SESSION_CWD.set(wd) logger.info("[%s] 设置 agent 工作目录 contextvar: %s", run_id, wd) except Exception as e: logger.warning("[%s] 设置工作目录失败（agent 将使用进程 cwd）: %s", run_id, e) try: agent.tool_start_callback = on_tool_start agent.tool_complete_callback = on_tool_end result = agent.run_conversation( user_message=user_message, system_message=system_prompt, stream_callback=on_stream, task_id=effective_task_id, ) final = result.get("final_response", "") if isinstance(result, dict) else str(result) queue.append(("done", {"content": final})) except Exception as e: logger.error("Agent 执行异常: %s", e, exc_info=True) queue.append(("error", {"message": str(e)})) finally: if cwd_token is not None: try: _SESSION_CWD.reset(cwd_token) except Exception: pass if original_cwd is not None: try: os.chdir(original_cwd) except Exception: pass done_event.set() thread = threading.Thread(target=_run, daemon=True) thread.start() while not done_event.is_set() or queue: while queue: event_type, data = queue.pop(0) yield f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" time.sleep(0.05) # 刷出最后可能残留的事件 while queue: event_type, data = queue.pop(0) yield f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" # --------------------------------------------------------------------------- # API 端点 # --------------------------------------------------------------------------- @app.post("/run") async def run_agent(req: RunRequest): """ 运行 Hermes Agent，返回 SSE 流式响应。 事件类型： - text: 流式文本片段 - tool_start: 工具调用开始 - tool_end: 工具调用结束 - done: 执行完成，包含最终回复 - error: 执行出错 """ if not req.user_message.strip(): raise HTTPException(status_code=400, detail="user_message 不能为空") session_id = req.session_id or "" run_id = uuid.uuid4().hex[:8] try: agent = _get_or_create_agent( session_id, req.max_iterations, req.hermes_home, model_id=req.model_id, base_url=req.base_url, api_key=req.api_key, model_name=req.model_name ) except (ImportError, ValueError) as e: raise HTTPException(status_code=503, detail=str(e)) return StreamingResponse( _run_agent_stream(agent, req.system_prompt, req.user_message, run_id, req.working_dir, session_id), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive", }, ) @app.get("/health") async def health(): """健康检查""" hermes_home = os.environ.get("HERMES_HOME", str(Path.home() / ".hermes")) hermes_agent_found = HERMES_AGENT_DIR.is_dir() # 尝试导入验证 can_import = False try: import run_agent can_import = True except ImportError: pass return { "status": "ok" if (hermes_agent_found or can_import) else "degraded", "hermes_home": hermes_home, "hermes_agent_dir": str(HERMES_AGENT_DIR), "hermes_agent_found": hermes_agent_found, "can_import_agent": can_import, "active_sessions": len(_agents), } # --------------------------------------------------------------------------- # 启动 # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Hermes Bridge Server") parser.add_argument("--port", type=int, default=18731, help="监听端口 (默认 18731)") parser.add_argument("--host", default="127.0.0.1", help="监听地址 (默认 127.0.0.1)") parser.add_argument("--hermes-home", default=None, help="Hermes Home 目录") args = parser.parse_args() if args.hermes_home: os.environ["HERMES_HOME"] = args.hermes_home import uvicorn logger.info("启动 Hermes Bridge: %s:%d", args.host, args.port) uvicorn.run(app, host=args.host, port=args.port, log_level="info") if __name__ == "__main__": main()