查询 LLM 并解析 JSON 响应
(self, prompt: str)
| 118 | log(f"AHP Server Agent 已就绪 (config={config}, skills={skill_dir})") |
| 119 | |
| 120 | def _llm_decide(self, prompt: str) -> Dict[str, Any]: |
| 121 | """查询 LLM 并解析 JSON 响应""" |
| 122 | self._ensure_session() |
| 123 | try: |
| 124 | result = self.session.send(prompt) |
| 125 | text = result.text.strip() |
| 126 | start = text.find("{") |
| 127 | end = text.rfind("}") + 1 |
| 128 | if start >= 0 and end > start: |
| 129 | return json.loads(text[start:end]) |
| 130 | log(f"响应中未找到 JSON: {text[:100]}") |
| 131 | except Exception as e: |
| 132 | log(f"LLM 错误: {e}") |
| 133 | # 解析失败时的保守回退 |
| 134 | return {"action": "continue", "reason": "分析失败,默认允许"} |
| 135 | |
| 136 | def _handle_pre_tool_use(self, payload: Dict[str, Any], depth: int) -> Dict[str, Any]: |
| 137 | tool = payload.get("tool", "unknown") |
no test coverage detected