diff --git a/backend/agent_tools.py b/backend/agent_tools.py index 48aef4a..0bf995f 100644 --- a/backend/agent_tools.py +++ b/backend/agent_tools.py @@ -7,7 +7,7 @@ from backend.diagnosis import KUBECTL READ_ONLY_VERBS = { - "api-resources", "api-versions", "auth", "cluster-info", "describe", + "api-resources", "api-versions", "auth", "cluster-info", "config", "describe", "diff", "explain", "get", "logs", "top", "version", "wait", } WRITE_OR_INTERACTIVE_VERBS = { @@ -69,9 +69,21 @@ def _find_verb(args: list[str]) -> str: raise ValueError("无法识别 kubectl 操作类型") +READ_ONLY_CONFIG_SUBCOMMANDS = {"current-context", "get-clusters", "get-contexts", "get-users", "view"} + + def classify_kubectl_command(command: str) -> CommandDecision: args = parse_kubectl_command(command) verb = _find_verb(args) + if verb == "config": + try: + verb_index = args.index(next(arg for arg in args if arg.lower() == "config")) + subcommand = args[verb_index + 1].lower() + except (StopIteration, ValueError, IndexError): + return CommandDecision("write", True, verb, args) + if subcommand in READ_ONLY_CONFIG_SUBCOMMANDS: + return CommandDecision("read", False, verb, args) + return CommandDecision("write", True, verb, args) if verb in READ_ONLY_VERBS: return CommandDecision("read", False, verb, args) if verb in WRITE_OR_INTERACTIVE_VERBS: diff --git a/backend/ai_agent.py b/backend/ai_agent.py index c8a42ef..02e8843 100644 --- a/backend/ai_agent.py +++ b/backend/ai_agent.py @@ -3,6 +3,7 @@ import json import re import uuid from collections import Counter +from typing import Optional import httpx from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL from backend.agent_tools import classify_kubectl_command, execute_kubectl @@ -40,7 +41,9 @@ AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职 每轮必须只返回一个 JSON 对象,不要使用 Markdown 代码块: - 继续调查:{"action":"command","command":"kubectl ...","reason":"为什么执行"} -- 完成分析:{"action":"final","analysis":"中文 Markdown 分析结论"} +- 完成分析:{"action":"final","analysis":"中文 Markdown 分析结论","verified":true,"remaining_issues":0} + +只有完成修复后的全量复检,并确认 remaining_issues 为 0 时才能返回 final。只定位根因、执行命令成功或部分资源恢复都不能结束。 """ _PENDING_APPROVALS: dict[str, dict] = {} @@ -72,6 +75,9 @@ def _parse_agent_action(content: str) -> dict: raise ValueError("AI 命令动作缺少 command") if action["action"] == "final" and not action.get("analysis"): raise ValueError("AI 最终动作缺少 analysis") + if action["action"] == "final": + if action.get("verified") is not True or action.get("remaining_issues") != 0: + raise ValueError("AI 最终动作必须确认 verified=true 且 remaining_issues=0") return action @@ -130,17 +136,58 @@ async def run_diagnostic_agent( "content": "AI 执行模式已自动采集第一份现场证据:\n" + json.dumps(initial_result, ensure_ascii=False), }) - for step in range(1, max_steps + 1): + async for event in _run_agent_loop( + messages=messages, + execution_mode=execution_mode, + max_steps=max_steps, + start_step=1, + command_counts=command_counts, + report=report, + question=question, + ): + yield event + + +async def _run_agent_loop( + messages: list[dict], + execution_mode: str, + max_steps: int, + start_step: int, + command_counts: Counter, + report: str, + question: str, +): + for step in range(start_step, max_steps + 1): yield {"type": "thinking", "step": step, "message": f"AI 正在规划第 {step} 轮诊断"} try: content = await _complete_chat(messages) action = _parse_agent_action(content) + except ValueError as exc: + if "最终动作必须确认" in str(exc): + messages.extend([ + {"role": "assistant", "content": content}, + { + "role": "user", + "content": ( + "不能结束诊断:尚未通过修复后的全量复检确认所有问题清零。" + "请继续返回 command,检查节点、Pod、Deployment、事件以及原故障资源;" + "只有 verified=true 且 remaining_issues=0 时才可返回 final。" + ), + }, + ]) + yield {"type": "verification_required", "message": "尚未确认所有问题清零,Agent 继续执行诊断验证"} + continue + yield {"type": "error", "message": str(exc)} + return except Exception as exc: yield {"type": "error", "message": str(exc)} return if action["action"] == "final": - yield {"type": "final", "content": action["analysis"], "model": AI_MODEL} + yield { + "type": "final", "content": action["analysis"], "model": AI_MODEL, + "verified": True, "remaining_issues": 0, + } return command = action["command"] @@ -191,6 +238,13 @@ async def run_diagnostic_agent( "reason": reason, "created_by": AI_MODEL, "run_id": "", + "report": report, + "question": question, + "execution_mode": execution_mode, + "messages": messages + [{"role": "assistant", "content": content}], + "next_step": step + 1, + "max_steps": max_steps, + "command_counts": dict(command_counts), } yield { "type": "approval_required", "approval_id": approval_id, @@ -214,6 +268,28 @@ async def run_diagnostic_agent( } +async def resume_diagnostic_agent(pending: dict, result: dict): + """审批命令执行后,把真实结果交回原 Agent 上下文并继续诊断。""" + messages = list(pending.get("messages") or []) + messages.append({ + "role": "user", + "content": ( + "用户已批准命令,以下是实际执行结果。请继续自主诊断并验证问题是否解决:\n" + + json.dumps(result, ensure_ascii=False) + ), + }) + async for event in _run_agent_loop( + messages=messages, + execution_mode=pending.get("execution_mode", "ai"), + max_steps=int(pending.get("max_steps", 30)), + start_step=int(pending.get("next_step", 1)), + command_counts=Counter(pending.get("command_counts") or {}), + report=pending.get("report", ""), + question=pending.get("question", ""), + ): + yield event + + def take_pending_approval(approval_id: str) -> dict | None: return _PENDING_APPROVALS.pop(approval_id, None) diff --git a/backend/main.py b/backend/main.py index dedd08a..a7c423e 100644 --- a/backend/main.py +++ b/backend/main.py @@ -22,7 +22,7 @@ from backend.diagnosis import ( from backend.ai_agent import ( analyze_with_ai, chat_with_ai, analyze_with_ai_stream, chat_with_ai_stream, - run_diagnostic_agent, take_pending_approval, + run_diagnostic_agent, resume_diagnostic_agent, take_pending_approval, ) from backend.agent_tools import execute_kubectl from backend.history_store import HistoryStore @@ -371,12 +371,41 @@ async def execute_approved_agent_command(req: AgentApprovalRequest): history_store.finish_run(pending["run_id"], "rejected", result["message"]) return result result = await execute_kubectl(pending["command"]) - response = {"approved": True, "reason": pending["reason"], **result} - if pending.get("run_id"): - history_store.append_event(pending["run_id"], "approved_result", response) - status = "completed" if result.get("returncode") == 0 else "failed" - history_store.finish_run(pending["run_id"], status, result.get("stdout") or result.get("stderr", "")) - return response + approved_event = {"approved": True, "reason": pending["reason"], **result} + + async def event_generator(): + yield f"data: {json.dumps({'type': 'approved_result', **approved_event}, ensure_ascii=False)}\n\n" + run_id = pending.get("run_id", "") + if run_id: + history_store.append_event(run_id, "approved_result", approved_event) + + final_status = "completed" + summary = result.get("stdout") or result.get("stderr", "") + async for event in resume_diagnostic_agent(pending, result): + if run_id: + history_store.append_event(run_id, event.get("type", "event"), {k: v for k, v in event.items() if k != "type"}) + if event.get("type") == "final": + summary = event.get("content", "") + elif event.get("type") == "error": + final_status = "failed" + summary = event.get("message", "") + elif event.get("type") == "approval_required": + final_status = "waiting" + from backend.ai_agent import _PENDING_APPROVALS + next_pending = _PENDING_APPROVALS.get(event.get("approval_id")) + if next_pending is not None: + next_pending["run_id"] = run_id + yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" + + if run_id: + history_store.finish_run(run_id, final_status, summary) + yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n" + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) @app.post("/api/chat") diff --git a/frontend/src/App.vue b/frontend/src/App.vue index a0cdc89..9e78eeb 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -712,21 +712,51 @@ async function resolveApproval(event, approved) { headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ approval_id: event.approval_id, approved }), }) - const result = await resp.json() - if (!resp.ok) throw new Error(result.detail || `HTTP ${resp.status}`) - event.resolved = true - event.approvalSuccess = approved && result.returncode === 0 - if (!approved) { - event.approvalResult = '已拒绝执行该命令' - } else { - event.approvalResult = result.returncode === 0 ? '命令执行成功' : `命令执行失败:${result.stderr || result.returncode}` - agentEvents.value.push({ type: 'result', ...result }) + if (!resp.ok) { + const error = await resp.json() + throw new Error(error.detail || `HTTP ${resp.status}`) } - ElMessage[approved && result.returncode === 0 ? 'success' : 'info'](event.approvalResult) + + if (!approved) { + const result = await resp.json() + event.resolved = true + event.approvalSuccess = false + event.approvalResult = '已拒绝执行该命令' + ElMessage.info(event.approvalResult) + return + } + + if (!resp.body) throw new Error('审批接口未返回流式响应') + event.resolved = true + agentRunning.value = true + const reader = resp.body.getReader() + const decoder = new TextDecoder() + let buffer = '' + while (true) { + const { done, value } = await reader.read() + if (done) break + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() || '' + for (const line of lines) { + if (!line.startsWith('data: ')) continue + const nextEvent = JSON.parse(line.slice(6)) + if (nextEvent.type === 'approved_result') { + event.approvalSuccess = nextEvent.returncode === 0 + event.approvalResult = nextEvent.returncode === 0 ? '命令执行成功,Agent 继续诊断中' : `命令执行失败:${nextEvent.stderr || nextEvent.returncode}` + agentEvents.value.push({ type: 'result', ...nextEvent }) + } else if (!['start', 'done'].includes(nextEvent.type)) { + agentEvents.value.push(nextEvent) + if (nextEvent.type === 'error') ElMessage.error(nextEvent.message) + } + } + } + ElMessage.success('审批命令已执行,Agent 已继续诊断') } catch (e) { ElMessage.error('审批处理失败: ' + e.message) } finally { event.executing = false + agentRunning.value = false } } diff --git a/tests/test_agent_api.py b/tests/test_agent_api.py index ebc14f6..191ace1 100644 --- a/tests/test_agent_api.py +++ b/tests/test_agent_api.py @@ -29,19 +29,68 @@ class AgentApiTest(unittest.TestCase): self.assertEqual(events[-1]["type"], "done") self.assertEqual(events[-2]["type"], "final") + def test_config_view_is_read_only_and_does_not_require_approval(self): + from backend.agent_tools import classify_kubectl_command + + decision = classify_kubectl_command("kubectl config view") + + self.assertEqual(decision.mode, "read") + self.assertFalse(decision.requires_approval) + + def test_execute_approved_command_continues_agent_with_command_result(self): + approval = { + "command": "kubectl rollout restart deployment/api -n prod", + "reason": "重启工作负载", + "run_id": "", + "report": "diagnosis report", + "question": "修复 API", + "execution_mode": "ai", + "messages": [{"role": "system", "content": "context"}], + "next_step": 2, + "max_steps": 30, + "command_counts": {"kubectl rollout restart deployment/api -n prod": 1}, + } + + async def fake_execute(command): + return {"command": command, "mode": "write", "returncode": 0, "stdout": "restarted", "stderr": ""} + + async def fake_resume(pending, result): + self.assertEqual(result["stdout"], "restarted") + yield {"type": "thinking", "step": 2, "message": "继续验证"} + yield {"type": "command", "command": "kubectl get pods -n prod", "mode": "read", "reason": "验证恢复"} + yield {"type": "result", "command": "kubectl get pods -n prod", "mode": "read", "returncode": 0, "stdout": "Running", "stderr": ""} + yield {"type": "final", "content": "故障已恢复", "model": "test"} + + async def run(): + with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_kubectl", fake_execute), patch.object(main, "resume_diagnostic_agent", fake_resume): + response = await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True)) + return [parse_sse(event) async for event in response.body_iterator] + + events = asyncio.run(run()) + self.assertEqual(events[0]["type"], "approved_result") + self.assertEqual(events[1]["type"], "thinking") + self.assertEqual(events[-2]["type"], "final") + self.assertEqual(events[-1]["type"], "done") + def test_execute_approved_command_once(self): approval = {"command": "kubectl scale deployment/api --replicas=2", "reason": "恢复副本"} async def fake_execute(command): return {"command": command, "mode": "write", "returncode": 0, "stdout": "scaled", "stderr": ""} - async def run(): - with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_kubectl", fake_execute): - return await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True)) + async def fake_resume(pending, result): + yield {"type": "final", "content": "验证完成", "model": "test"} - result = asyncio.run(run()) - self.assertEqual(result["returncode"], 0) - self.assertEqual(result["stdout"], "scaled") + async def run(): + with patch.object(main, "take_pending_approval", return_value=approval), patch.object(main, "execute_kubectl", fake_execute), patch.object(main, "resume_diagnostic_agent", fake_resume): + response = await main.execute_approved_agent_command(main.AgentApprovalRequest(approval_id="abc", approved=True)) + return [parse_sse(event) async for event in response.body_iterator] + + events = asyncio.run(run()) + self.assertEqual(events[0]["type"], "approved_result") + self.assertEqual(events[0]["returncode"], 0) + self.assertEqual(events[0]["stdout"], "scaled") + self.assertEqual(events[-1]["type"], "done") if __name__ == "__main__": diff --git a/tests/test_diagnostic_agent.py b/tests/test_diagnostic_agent.py index 8a73862..2b0fa17 100644 --- a/tests/test_diagnostic_agent.py +++ b/tests/test_diagnostic_agent.py @@ -13,7 +13,7 @@ class DiagnosticAgentTest(unittest.TestCase): def test_executes_read_only_tool_and_returns_final_answer(self): responses = iter([ '{"action":"command","command":"kubectl get pods -A","reason":"查看异常 Pod"}', - '{"action":"final","analysis":"发现一个异常 Pod"}', + '{"action":"final","analysis":"发现一个异常 Pod","verified":true,"remaining_issues":0}', ]) async def fake_completion(messages): @@ -90,7 +90,7 @@ class DiagnosticAgentTest(unittest.TestCase): def test_ai_mode_forces_initial_read_only_probe_before_model_planning(self): responses = iter([ - '{"action":"final","analysis":"根据采集结果完成分析"}', + '{"action":"final","analysis":"根据采集结果完成分析","verified":true,"remaining_issues":0}', ]) async def fake_completion(messages): @@ -131,7 +131,7 @@ class DiagnosticAgentTest(unittest.TestCase): ) for index in range(command_count) ], - '{"action":"final","analysis":"已完成根因分析"}', + '{"action":"final","analysis":"已完成根因分析","verified":true,"remaining_issues":0}', ]) executed = [] @@ -168,6 +168,44 @@ class DiagnosticAgentTest(unittest.TestCase): self.assertIn("安全上限", events[-1]["message"]) self.assertNotIn("人工排查", events[-1]["message"]) + def test_agent_rejects_final_until_verification_reports_no_remaining_issues(self): + responses = iter([ + '{"action":"final","analysis":"已经修复"}', + '{"action":"command","command":"kubectl get pods -A","reason":"执行全量复检"}', + '{"action":"final","analysis":"所有工作负载恢复正常","verified":true,"remaining_issues":0}', + ]) + executed = [] + + async def fake_completion(messages): + return next(responses) + + async def fake_execute(command): + executed.append(command) + stdout = "node-a Ready" if command == "kubectl get nodes -o wide" else "all pods Running" + return {"command": command, "mode": "read", "returncode": 0, "stdout": stdout, "stderr": ""} + + async def run(): + with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_kubectl", fake_execute): + return await collect_events(run_diagnostic_agent("report", execution_mode="ai", max_steps=5)) + + events = asyncio.run(run()) + self.assertEqual(executed, ["kubectl get nodes -o wide", "kubectl get pods -A"]) + self.assertEqual(events[-1]["type"], "final") + self.assertTrue(events[-1]["verified"]) + self.assertEqual(events[-1]["remaining_issues"], 0) + + def test_parse_final_requires_explicit_verification_fields(self): + from backend.ai_agent import _parse_agent_action + + with self.assertRaises(ValueError): + _parse_agent_action('{"action":"final","analysis":"看起来正常"}') + + action = _parse_agent_action( + '{"action":"final","analysis":"全部正常","verified":true,"remaining_issues":0}' + ) + self.assertTrue(action["verified"]) + self.assertEqual(action["remaining_issues"], 0) + def test_stops_when_model_returns_invalid_action(self): async def fake_completion(messages): return "not-json"