Class AHPServerAgent

examples/ahp_agent_monitors_agent.py:95–230

使用 A3S Code + Kimi 的 LLM 驱动 AHP 服务器

class AHPServerAgent:
    """使用 A3S Code + Kimi 的 LLM 驱动 AHP 服务器"""

    def __init__(self):
        self.session = None
        self.stats = {"requests": 0, "allowed": 0, "blocked": 0}

    def _ensure_session(self):
        """懒初始化：第一次请求时才创建 session（避免启动超时）"""
        if self.session is not None:
            return
        config = find_config()
        agent = Agent.create(config)
        import tempfile
        workspace = tempfile.mkdtemp(prefix="ahp_srv_")

        # 挂载 ahp_skills 目录，让 AHP Server 智能体使用 skill 分析工具调用
        skill_dir = str(Path(__file__).parent / "ahp_skills")
        opts = SessionOptions()
        opts.builtin_skills = False
        opts.skill_dirs = [skill_dir]
        opts.permission_policy = PermissionPolicy(default_decision="allow")
        self.session = agent.session(workspace, opts)
        log(f"AHP Server Agent 已就绪 (config={config}, skills={skill_dir})")

    def _llm_decide(self, prompt: str) -> Dict[str, Any]:
        """查询 LLM 并解析 JSON 响应"""
        self._ensure_session()
        try:
            result = self.session.send(prompt)
            text = result.text.strip()
            start = text.find("{")
            end = text.rfind("}") + 1
            if start >= 0 and end > start:
                return json.loads(text[start:end])
            log(f"响应中未找到 JSON: {text[:100]}")
        except Exception as e:
            log(f"LLM 错误: {e}")
        # 解析失败时的保守回退
        return {"action": "continue", "reason": "分析失败，默认允许"}

    def _handle_pre_tool_use(self, payload: Dict[str, Any], depth: int) -> Dict[str, Any]:
        tool = payload.get("tool", "unknown")
        args = payload.get("arguments", {})
        self.stats["requests"] += 1
        log(f"pre_tool_use: tool={tool} args={json.dumps(args)[:80]} depth={depth}")

        prompt = PRE_TOOL_PROMPT.format(
            tool=tool,
            args=json.dumps(args, ensure_ascii=False),
            depth=depth,
        )
        decision = self._llm_decide(prompt)
        action = decision.get("action", "continue")
        reason = decision.get("reason", "")

        if action == "block":
            self.stats["blocked"] += 1