diff --git a/.gitignore b/.gitignore index 4a3d4a1..97b896d 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,12 @@ venv/ # Environment .env +# Runtime data +data/ +*.db +*.db-shm +*.db-wal + # IDE .vscode/ .idea/ diff --git a/README.md b/README.md index c97ebec..260ebdf 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,11 @@ - 实时展示思考步骤、执行命令和真实输出 - 即使选择 AI 执行，`apply`、`delete`、`patch`、`scale`、`rollout`、`exec` 等命令仍必须人工确认 - 禁止 shell、管道、重定向和多命令拼接 +- **历史记录持久化**：使用 SQLite 保存全部诊断与处理轨迹 + - 保存完整诊断结果、文本报告、AI 分析结果 + - 保存 Agent 思考、命令、真实输出、审批和执行结果 + - 支持按状态、命名空间查询并查看完整详情 + - 服务重启后仍可读取历史和恢复最近一次诊断 - **AI 追问对话**：基于诊断上下文多轮追问，深入排查问题 - **Web 仪表盘**：Vue3 + Element Plus，实时展示诊断结果 @@ -50,6 +55,9 @@ PORT=8900 # Kubeconfig 路径 (留空使用默认 ~/.kube/config) KUBECONFIG_PATH= + +# 历史记录 SQLite 文件（K8S 部署时请挂载持久卷到该路径） +HISTORY_DB_PATH=/app/data/history.db ``` ## API 接口 @@ -63,6 +71,8 @@ KUBECONFIG_PATH= | POST | `/api/analyze` | AI 分析诊断结果 `{"question":""}` | | GET | `/api/agent/diagnose/stream` | 自主诊断 SSE 流，`execution_mode=manual|ai` | | POST | `/api/agent/approve` | 批准或拒绝待执行命令 `{"approval_id":"...","approved":true}` | +| GET | `/api/history/diagnoses` | 查询全部诊断历史，支持 `status`、`namespace` 过滤 | +| GET | `/api/history/diagnoses/{id}` | 查询诊断详情及全部 AI/Agent 处理轨迹 | | POST | `/api/chat` | AI 多轮对话 `{"messages":[...]}` | | GET | `/api/report` | 获取诊断报告文本 | diff --git a/backend/ai_agent.py b/backend/ai_agent.py index 2f1b879..dee5add 100644 --- a/backend/ai_agent.py +++ b/backend/ai_agent.py @@ -1,5 +1,6 @@ """AI Agent 分析模块 - 调用 OpenAI 兼容接口分析诊断结果 (支持流式输出)""" import json +import re import uuid import httpx from backend.config import AI_API_BASE, AI_API_KEY, AI_MODEL @@ -22,7 +23,12 @@ _HEADERS = { } -AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你可以根据诊断报告决定执行 kubectl 命令获取更多证据，并在证据充分后给出结论。 +AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职责是主动调用 kubectl 获取现场证据，而不是只给用户一段脚本或操作建议。 + +行为要求： +1. 如果证据不足，必须返回 command 动作并逐条调查；至少检查与故障资源直接相关的状态、describe、events 或 logs。 +2. 禁止在 final 中仅提供“请执行以下命令”的脚本。只有在已经依据真实命令输出形成结论时才能返回 final。 +3. 每次只提出一条 kubectl 命令，服务端会执行并把真实输出返回给你，然后你继续下一轮判断。 安全规则： 1. 只允许提出单条 kubectl 命令，不得使用 shell、管道、重定向或命令连接符。 @@ -40,10 +46,25 @@ _PENDING_APPROVALS: dict[str, dict] = {} def _parse_agent_action(content: str) -> dict: - try: - action = json.loads(content.strip()) - except json.JSONDecodeError as exc: - raise ValueError("AI 未返回合法 JSON 动作") from exc + text = content.strip() + candidates = [text] + fenced = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL | re.IGNORECASE) + if fenced: + candidates.insert(0, fenced.group(1)) + first_brace = text.find("{") + last_brace = text.rfind("}") + if first_brace >= 0 and last_brace > first_brace: + candidates.append(text[first_brace:last_brace + 1]) + + action = None + for candidate in candidates: + try: + action = json.loads(candidate) + break + except json.JSONDecodeError: + continue + if action is None: + raise ValueError("AI 未返回合法 JSON 动作") if not isinstance(action, dict) or action.get("action") not in {"command", "final"}: raise ValueError("AI 返回了不支持的动作") if action["action"] == "command" and not action.get("command"): @@ -91,6 +112,20 @@ async def run_diagnostic_agent( {"role": "user", "content": user_content}, ] + if execution_mode == "ai": + initial_command = "kubectl get nodes -o wide" + yield { + "type": "command", "step": 0, "command": initial_command, + "reason": "建立集群节点状态基线，确保 AI 执行模式实际采集现场证据", + "mode": "read", + } + initial_result = await execute_kubectl(initial_command) + yield {"type": "result", **initial_result} + messages.append({ + "role": "user", + "content": "AI 执行模式已自动采集第一份现场证据：\n" + json.dumps(initial_result, ensure_ascii=False), + }) + for step in range(1, max_steps + 1): yield {"type": "thinking", "step": step, "message": f"AI 正在规划第 {step} 轮诊断"} try: @@ -134,6 +169,7 @@ async def run_diagnostic_agent( "command": command, "reason": reason, "created_by": AI_MODEL, + "run_id": "", } yield { "type": "approval_required", "approval_id": approval_id, diff --git a/backend/config.py b/backend/config.py index 055f292..4b51498 100644 --- a/backend/config.py +++ b/backend/config.py @@ -9,3 +9,4 @@ AI_API_KEY = os.getenv("AI_API_KEY", "hermes") AI_MODEL = os.getenv("AI_MODEL", "qwen3.8-max-preview") PORT = int(os.getenv("PORT", "8900")) KUBECONFIG_PATH = os.getenv("KUBECONFIG_PATH", "") or None +HISTORY_DB_PATH = os.getenv("HISTORY_DB_PATH", os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "history.db")) diff --git a/backend/history_store.py b/backend/history_store.py new file mode 100644 index 0000000..ca6e7b0 --- /dev/null +++ b/backend/history_store.py @@ -0,0 +1,172 @@ +"""SQLite 持久化诊断历史、AI 分析和 Agent 处理记录。""" +import json +import os +import sqlite3 +import uuid +from datetime import datetime +from typing import Optional + + +def _now() -> str: + return datetime.now().isoformat(timespec="seconds") + + +class HistoryStore: + def __init__(self, path: str): + self.path = path + parent = os.path.dirname(os.path.abspath(path)) + os.makedirs(parent, exist_ok=True) + self._init_db() + + def _connect(self): + connection = sqlite3.connect(self.path, timeout=30) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA foreign_keys=ON") + return connection + + def _init_db(self): + with self._connect() as db: + db.executescript(""" + CREATE TABLE IF NOT EXISTS diagnoses ( + id TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + namespace TEXT NOT NULL, + score INTEGER NOT NULL, + status TEXT NOT NULL, + critical_count INTEGER NOT NULL DEFAULT 0, + warning_count INTEGER NOT NULL DEFAULT 0, + elapsed_seconds REAL NOT NULL DEFAULT 0, + diagnosis_json TEXT NOT NULL, + report TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_diagnoses_created_at ON diagnoses(created_at DESC); + CREATE INDEX IF NOT EXISTS idx_diagnoses_status ON diagnoses(status); + + CREATE TABLE IF NOT EXISTS processing_runs ( + id TEXT PRIMARY KEY, + diagnosis_id TEXT NOT NULL, + kind TEXT NOT NULL, + mode TEXT NOT NULL DEFAULT '', + question TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL, + summary TEXT NOT NULL DEFAULT '', + started_at TEXT NOT NULL, + finished_at TEXT, + FOREIGN KEY(diagnosis_id) REFERENCES diagnoses(id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_runs_diagnosis ON processing_runs(diagnosis_id, started_at); + + CREATE TABLE IF NOT EXISTS processing_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + run_id TEXT NOT NULL, + sequence INTEGER NOT NULL, + type TEXT NOT NULL, + created_at TEXT NOT NULL, + payload_json TEXT NOT NULL, + FOREIGN KEY(run_id) REFERENCES processing_runs(id) ON DELETE CASCADE, + UNIQUE(run_id, sequence) + ); + """) + + def save_diagnosis(self, diagnosis: dict, report: str) -> str: + record_id = uuid.uuid4().hex + created_at = diagnosis.get("timestamp") or _now() + with self._connect() as db: + db.execute( + """INSERT INTO diagnoses + (id, created_at, namespace, score, status, critical_count, warning_count, + elapsed_seconds, diagnosis_json, report) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + record_id, created_at, diagnosis.get("namespace", "all"), + int(diagnosis.get("score", 0)), diagnosis.get("status", "unknown"), + int(diagnosis.get("critical_count", 0)), int(diagnosis.get("warning_count", 0)), + float(diagnosis.get("elapsed_seconds", 0)), + json.dumps(diagnosis, ensure_ascii=False), report, + ), + ) + return record_id + + def list_diagnoses(self, status: str = "", namespace: str = "", limit: int = 100, offset: int = 0) -> list[dict]: + conditions = [] + params: list = [] + if status: + conditions.append("status = ?") + params.append(status) + if namespace: + conditions.append("namespace = ?") + params.append(namespace) + where = " WHERE " + " AND ".join(conditions) if conditions else "" + params.extend([max(1, min(limit, 500)), max(0, offset)]) + with self._connect() as db: + rows = db.execute( + f"""SELECT id, created_at, namespace, score, status, critical_count, + warning_count, elapsed_seconds, + (SELECT COUNT(*) FROM processing_runs r WHERE r.diagnosis_id = diagnoses.id) AS run_count + FROM diagnoses{where} + ORDER BY created_at DESC, rowid DESC LIMIT ? OFFSET ?""", + params, + ).fetchall() + return [dict(row) for row in rows] + + def get_diagnosis(self, record_id: str) -> Optional[dict]: + with self._connect() as db: + row = db.execute("SELECT * FROM diagnoses WHERE id = ?", (record_id,)).fetchone() + if row is None: + return None + run_rows = db.execute( + "SELECT * FROM processing_runs WHERE diagnosis_id = ? ORDER BY started_at, rowid", + (record_id,), + ).fetchall() + runs = [] + for run_row in run_rows: + run = dict(run_row) + event_rows = db.execute( + "SELECT type, created_at, payload_json FROM processing_events WHERE run_id = ? ORDER BY sequence", + (run["id"],), + ).fetchall() + run["events"] = [ + {"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])} + for event in event_rows + ] + runs.append(run) + return { + "id": row["id"], "created_at": row["created_at"], + "diagnosis": json.loads(row["diagnosis_json"]), + "report": row["report"], "runs": runs, + } + + def latest(self) -> Optional[dict]: + records = self.list_diagnoses(limit=1) + return self.get_diagnosis(records[0]["id"]) if records else None + + def create_run(self, diagnosis_id: str, kind: str, mode: str = "", question: str = "") -> str: + run_id = uuid.uuid4().hex + with self._connect() as db: + db.execute( + """INSERT INTO processing_runs + (id, diagnosis_id, kind, mode, question, status, started_at) + VALUES (?, ?, ?, ?, ?, 'running', ?)""", + (run_id, diagnosis_id, kind, mode, question, _now()), + ) + return run_id + + def append_event(self, run_id: str, event_type: str, payload: dict): + with self._connect() as db: + sequence = db.execute( + "SELECT COALESCE(MAX(sequence), 0) + 1 FROM processing_events WHERE run_id = ?", + (run_id,), + ).fetchone()[0] + db.execute( + """INSERT INTO processing_events (run_id, sequence, type, created_at, payload_json) + VALUES (?, ?, ?, ?, ?)""", + (run_id, sequence, event_type, _now(), json.dumps(payload, ensure_ascii=False)), + ) + + def finish_run(self, run_id: str, status: str, summary: str = ""): + with self._connect() as db: + db.execute( + "UPDATE processing_runs SET status = ?, summary = ?, finished_at = ? WHERE id = ?", + (status, summary, _now(), run_id), + ) diff --git a/backend/main.py b/backend/main.py index 4e49ee9..dedd08a 100644 --- a/backend/main.py +++ b/backend/main.py @@ -11,7 +11,7 @@ from fastapi.responses import StreamingResponse from pydantic import BaseModel from typing import Optional -from backend.config import PORT, AI_API_BASE, AI_MODEL +from backend.config import PORT, AI_API_BASE, AI_MODEL, HISTORY_DB_PATH from backend.diagnosis import ( run_full_diagnosis, build_diagnosis_report, ALL_CHECKS, check_connectivity, @@ -25,8 +25,10 @@ from backend.ai_agent import ( run_diagnostic_agent, take_pending_approval, ) from backend.agent_tools import execute_kubectl +from backend.history_store import HistoryStore app = FastAPI(title="K8S 智能诊断平台", version="1.0.0") +history_store = HistoryStore(HISTORY_DB_PATH) app.add_middleware( CORSMiddleware, @@ -38,6 +40,25 @@ app.add_middleware( # 存储最近诊断结果 (内存) _last_diagnosis: Optional[dict] = None _last_report: str = "" +_last_diagnosis_id: Optional[str] = None + + +def _save_current_diagnosis(diagnosis: dict, report: str) -> str: + global _last_diagnosis, _last_report, _last_diagnosis_id + _last_diagnosis = diagnosis + _last_report = report + _last_diagnosis_id = history_store.save_diagnosis(diagnosis, report) + return _last_diagnosis_id + + +def _ensure_current_history() -> str: + global _last_diagnosis_id + if _last_diagnosis_id: + return _last_diagnosis_id + if _last_diagnosis is None: + raise RuntimeError("暂无诊断记录") + _last_diagnosis_id = history_store.save_diagnosis(_last_diagnosis, _last_report) + return _last_diagnosis_id # 诊断项执行顺序 (带中文标题) CHECK_ORDER = [ @@ -121,9 +142,9 @@ async def diagnose_stream(namespace: str = ""): }], "checks": {}, } - _last_diagnosis = diagnosis - _last_report = build_diagnosis_report(diagnosis) - yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis}, ensure_ascii=False)}\n\n" + report = build_diagnosis_report(diagnosis) + _save_current_diagnosis(diagnosis, report) + yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n" return yield f"data: {json.dumps({'type': 'progress', 'step': 0, 'total': len(CHECK_ORDER) + 1, 'name': 'connectivity', 'title': '集群连通性检查', 'status': 'done', 'issues': 0}, ensure_ascii=False)}\n\n" @@ -176,10 +197,10 @@ async def diagnose_stream(namespace: str = ""): "checks": results, } - _last_diagnosis = diagnosis - _last_report = build_diagnosis_report(diagnosis) + report = build_diagnosis_report(diagnosis) + _save_current_diagnosis(diagnosis, report) - yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis}, ensure_ascii=False)}\n\n" + yield f"data: {json.dumps({'type': 'complete', 'diagnosis': diagnosis, 'history_id': _last_diagnosis_id}, ensure_ascii=False)}\n\n" return StreamingResponse( event_generator(), @@ -192,17 +213,24 @@ async def diagnose_stream(namespace: str = ""): async def diagnose(req: DiagnoseRequest): """执行一键诊断 (非流式，兼容旧接口)""" global _last_diagnosis, _last_report - _last_diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks) - _last_report = build_diagnosis_report(_last_diagnosis) - return _last_diagnosis + diagnosis = await run_full_diagnosis(namespace=req.namespace, checks=req.checks) + report = build_diagnosis_report(diagnosis) + _save_current_diagnosis(diagnosis, report) + return {**diagnosis, "history_id": _last_diagnosis_id} @app.get("/api/diagnosis/latest") async def get_latest(): - """获取最近一次诊断结果""" + """获取最近一次诊断结果，服务重启后从 SQLite 恢复。""" + global _last_diagnosis, _last_report, _last_diagnosis_id if _last_diagnosis is None: - raise HTTPException(status_code=404, detail="暂无诊断记录，请先执行诊断") - return _last_diagnosis + latest = history_store.latest() + if latest is None: + raise HTTPException(status_code=404, detail="暂无诊断记录，请先执行诊断") + _last_diagnosis = latest["diagnosis"] + _last_report = latest["report"] + _last_diagnosis_id = latest["id"] + return {**_last_diagnosis, "history_id": _last_diagnosis_id} @app.post("/api/analyze") @@ -210,19 +238,25 @@ async def analyze(req: AnalyzeRequest): """AI 分析诊断结果 (非流式，兼容旧接口)""" global _last_diagnosis, _last_report if _last_diagnosis is None: - _last_diagnosis = await run_full_diagnosis() - _last_report = build_diagnosis_report(_last_diagnosis) + diagnosis = await run_full_diagnosis() + report = build_diagnosis_report(diagnosis) + _save_current_diagnosis(diagnosis, report) + run_id = history_store.create_run(_ensure_current_history(), "analysis", "standard", req.question) result = await analyze_with_ai(_last_report, question=req.question) - return result + history_store.append_event(run_id, "analysis", result) + history_store.finish_run(run_id, "completed" if result.get("success") else "failed", result.get("analysis", "")) + return {**result, "run_id": run_id} @app.get("/api/analyze/stream") async def analyze_stream(question: str = ""): - """SSE 流式 AI 分析 - 逐字推送分析内容""" + """SSE 流式 AI 分析 - 逐字推送分析内容并持久化""" global _last_diagnosis, _last_report if _last_diagnosis is None: - _last_diagnosis = await run_full_diagnosis() - _last_report = build_diagnosis_report(_last_diagnosis) + diagnosis = await run_full_diagnosis() + report = build_diagnosis_report(diagnosis) + _save_current_diagnosis(diagnosis, report) + run_id = history_store.create_run(_ensure_current_history(), "analysis", "stream", question) async def event_generator(): process_steps = [ @@ -232,23 +266,39 @@ async def analyze_stream(question: str = ""): ("summary", "整理分析结论", "正在组织根因、影响和预防建议"), ] - yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL}, ensure_ascii=False)}\n\n" + yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'run_id': run_id}, ensure_ascii=False)}\n\n" + content = "" try: for step, title, detail in process_steps: - yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'running'}, ensure_ascii=False)}\n\n" + process_event = {'step': step, 'title': title, 'detail': detail, 'status': 'running'} + history_store.append_event(run_id, "process", process_event) + yield f"data: {json.dumps({'type': 'process', **process_event}, ensure_ascii=False)}\n\n" await asyncio.sleep(0) - yield f"data: {json.dumps({'type': 'process', 'step': step, 'title': title, 'detail': detail, 'status': 'done'}, ensure_ascii=False)}\n\n" + process_event = {'step': step, 'title': title, 'detail': detail, 'status': 'done'} + history_store.append_event(run_id, "process", process_event) + yield f"data: {json.dumps({'type': 'process', **process_event}, ensure_ascii=False)}\n\n" async for chunk in analyze_with_ai_stream(_last_report, question=question): + content += chunk yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n" - yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" + history_store.append_event(run_id, "final", {"content": content, "model": AI_MODEL}) + history_store.finish_run(run_id, "completed", content) + yield f"data: {json.dumps({'type': 'done', 'run_id': run_id}, ensure_ascii=False)}\n\n" except httpx.HTTPStatusError as e: err = f"AI 接口返回错误: {e.response.status_code} - {str(e)[:300]}" + history_store.append_event(run_id, "error", {"message": err}) + history_store.finish_run(run_id, "failed", err) yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n" except httpx.ConnectError: - yield f"data: {json.dumps({'type': 'error', 'message': f'无法连接 AI 服务 ({AI_API_BASE})，请检查 .env 中的 AI_API_BASE 配置'}, ensure_ascii=False)}\n\n" + err = f"无法连接 AI 服务 ({AI_API_BASE})，请检查 .env 中的 AI_API_BASE 配置" + history_store.append_event(run_id, "error", {"message": err}) + history_store.finish_run(run_id, "failed", err) + yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n" except Exception as e: - yield f"data: {json.dumps({'type': 'error', 'message': f'AI 分析失败: {str(e)}'}, ensure_ascii=False)}\n\n" + err = f"AI 分析失败: {str(e)}" + history_store.append_event(run_id, "error", {"message": err}) + history_store.finish_run(run_id, "failed", err) + yield f"data: {json.dumps({'type': 'error', 'message': err}, ensure_ascii=False)}\n\n" return StreamingResponse( event_generator(), @@ -264,21 +314,42 @@ async def agent_diagnose_stream(question: str = "", execution_mode: str = "ai"): raise HTTPException(status_code=400, detail="execution_mode 仅支持 ai 或 manual") global _last_diagnosis, _last_report if _last_diagnosis is None: - _last_diagnosis = await run_full_diagnosis() - _last_report = build_diagnosis_report(_last_diagnosis) + diagnosis = await run_full_diagnosis() + report = build_diagnosis_report(diagnosis) + _save_current_diagnosis(diagnosis, report) + run_id = history_store.create_run(_ensure_current_history(), "agent", execution_mode, question) async def event_generator(): - yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'mode': execution_mode}, ensure_ascii=False)}\n\n" + yield f"data: {json.dumps({'type': 'start', 'model': AI_MODEL, 'mode': execution_mode, 'run_id': run_id}, ensure_ascii=False)}\n\n" + final_status = "completed" + summary = "" try: async for event in run_diagnostic_agent( _last_report, question=question, execution_mode=execution_mode, ): + 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") in {"manual_command", "approval_required"}: + final_status = "waiting" + if event.get("type") == "approval_required": + from backend.ai_agent import _PENDING_APPROVALS + pending = _PENDING_APPROVALS.get(event.get("approval_id")) + if pending is not None: + pending["run_id"] = run_id yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n" - yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n" + 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" except Exception as exc: - yield f"data: {json.dumps({'type': 'error', 'message': f'自主诊断失败: {exc}'}, ensure_ascii=False)}\n\n" + message = f"自主诊断失败: {exc}" + history_store.append_event(run_id, "error", {"message": message}) + history_store.finish_run(run_id, "failed", message) + yield f"data: {json.dumps({'type': 'error', 'message': message}, ensure_ascii=False)}\n\n" return StreamingResponse( event_generator(), @@ -294,9 +365,18 @@ async def execute_approved_agent_command(req: AgentApprovalRequest): if pending is None: raise HTTPException(status_code=404, detail="审批请求不存在或已经处理") if not req.approved: - return {"approved": False, "command": pending["command"], "message": "用户已拒绝执行"} + result = {"approved": False, "command": pending["command"], "message": "用户已拒绝执行"} + if pending.get("run_id"): + history_store.append_event(pending["run_id"], "approval_rejected", result) + history_store.finish_run(pending["run_id"], "rejected", result["message"]) + return result result = await execute_kubectl(pending["command"]) - return {"approved": True, "reason": pending["reason"], **result} + 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 @app.post("/api/chat") @@ -325,6 +405,27 @@ async def chat_stream(req: ChatRequest): ) +@app.get("/api/history/diagnoses") +async def list_diagnosis_history( + status: str = "", + namespace: str = "", + limit: int = 100, + offset: int = 0, +): + """查询全部历史诊断记录，可按状态和命名空间过滤。""" + items = history_store.list_diagnoses(status=status, namespace=namespace, limit=limit, offset=offset) + return {"total": len(items), "items": items, "limit": limit, "offset": offset} + + +@app.get("/api/history/diagnoses/{record_id}") +async def get_diagnosis_history(record_id: str): + """查询单次诊断及其全部 AI 分析、Agent 命令和处理记录。""" + record = history_store.get_diagnosis(record_id) + if record is None: + raise HTTPException(status_code=404, detail="历史诊断记录不存在") + return record + + @app.get("/api/report") async def get_report(): """获取诊断报告文本""" diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 7ef6d39..a0cdc89 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -21,6 +21,7 @@ 手动执行命令 AI 执行命令 + 历史记录 {{ agentRunning ? 'Agent 诊断中...' : '自主诊断' }} @@ -337,6 +338,61 @@ + + +
+ + + + + + + 查询 +
+ + + + + + + + + + + + + + + +
+ 诊断详情 + + {{ historyDetail.created_at }} + {{ historyDetail.diagnosis.namespace }} + {{ historyDetail.diagnosis.score }} + {{ historyStatusText(historyDetail.diagnosis.status) }} + + + + 
{{ historyDetail.report }}
+ + + + + {{ run.mode || '-' }} + {{ run.question || '-' }} + {{ run.finished_at || '-' }} + +
+
{{ event.type }}{{ event.created_at }}
+ 
{{ formatHistoryEvent(event) }}
+
+ + +
+ @@ -359,6 +415,12 @@ const diagnosis = ref(null) const aiAnalysis = ref('') const aiModel = ref('') const analysisProcess = ref([]) +const historyVisible = ref(false) +const historyLoading = ref(false) +const historyItems = ref([]) +const historyDetail = ref(null) +const historyStatus = ref('') +const historyNamespace = ref('') const chatMessages = ref([]) const chatInput = ref('') const chatLoading = ref(false) @@ -529,6 +591,62 @@ async function runAnalysis() { } } +function openHistory() { + historyVisible.value = true +} + +async function loadHistory() { + historyLoading.value = true + historyDetail.value = null + try { + const params = new URLSearchParams({ limit: '200' }) + if (historyStatus.value) params.set('status', historyStatus.value) + if (historyNamespace.value) params.set('namespace', historyNamespace.value) + const resp = await fetch(`${API}/history/diagnoses?${params.toString()}`) + if (!resp.ok) throw new Error(`HTTP ${resp.status}`) + const data = await resp.json() + historyItems.value = data.items || [] + } catch (e) { + ElMessage.error('加载历史记录失败: ' + e.message) + } finally { + historyLoading.value = false + } +} + +async function loadHistoryDetail(row) { + historyLoading.value = true + try { + const resp = await fetch(`${API}/history/diagnoses/${row.id}`) + if (!resp.ok) throw new Error(`HTTP ${resp.status}`) + historyDetail.value = await resp.json() + } catch (e) { + ElMessage.error('加载历史详情失败: ' + e.message) + } finally { + historyLoading.value = false + } +} + +function historyStatusType(status) { + return status === 'healthy' ? 'success' : status === 'warning' ? 'warning' : status === 'critical' ? 'danger' : 'info' +} + +function historyStatusText(status) { + return { healthy: '健康', warning: '警告', critical: '严重' }[status] || status +} + +function runKindText(kind) { + return { analysis: 'AI 分析', agent: 'Agent 自主诊断', chat: 'AI 对话' }[kind] || kind +} + +function runStatusText(status) { + return { running: '运行中', completed: '已完成', waiting: '等待处理', failed: '失败', rejected: '已拒绝' }[status] || status +} + +function formatHistoryEvent(event) { + const { type, created_at, ...payload } = event + return JSON.stringify(payload, null, 2) +} + async function runAgentDiagnosis() { activeAgentMode.value = agentExecutionMode.value agentRunning.value = true @@ -781,6 +899,14 @@ body { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'PingFang SC' .manual-actions { display: flex; align-items: center; gap: 12px; margin-top: 12px; color: #606266; font-size: 13px; } .agent-running { display: flex; align-items: center; gap: 8px; color: #409eff; padding: 8px; } +.history-toolbar { display: flex; gap: 10px; margin-bottom: 16px; } +.history-detail { margin-top: 20px; } +.history-collapse { margin-top: 16px; } +.history-report { max-height: 480px; padding: 14px; overflow: auto; border-radius: 6px; background: #f4f4f5; white-space: pre-wrap; } +.history-event { margin-top: 10px; padding: 12px; border: 1px solid #ebeef5; border-radius: 6px; } +.history-event-head { display: flex; align-items: center; gap: 10px; color: #909399; font-size: 12px; } +.history-event pre { max-height: 300px; margin-top: 8px; padding: 10px; overflow: auto; background: #f4f4f5; white-space: pre-wrap; } + .chat-box { max-height: 400px; overflow-y: auto; padding: 12px 0; display: flex; flex-direction: column; gap: 12px; } .chat-msg { display: flex; } .chat-msg.user { justify-content: flex-end; } diff --git a/tests/test_diagnostic_agent.py b/tests/test_diagnostic_agent.py index c8bd020..ea6b622 100644 --- a/tests/test_diagnostic_agent.py +++ b/tests/test_diagnostic_agent.py @@ -27,8 +27,11 @@ class DiagnosticAgentTest(unittest.TestCase): return await collect_events(run_diagnostic_agent("report", max_steps=3)) events = asyncio.run(run()) - self.assertEqual([event["type"] for event in events], ["thinking", "command", "result", "thinking", "final"]) - self.assertEqual(events[1]["mode"], "read") + self.assertEqual( + [event["type"] for event in events], + ["command", "result", "thinking", "command", "result", "thinking", "final"], + ) + self.assertEqual(events[3]["mode"], "read") self.assertEqual(events[-1]["content"], "发现一个异常 Pod") def test_pauses_mutating_command_for_human_approval(self): @@ -63,13 +66,18 @@ class DiagnosticAgentTest(unittest.TestCase): async def fake_completion(messages): return '{"action":"command","command":"kubectl delete pod api-0 -n prod","reason":"删除故障 Pod"}' - async def run(): - with patch("backend.ai_agent._complete_chat", fake_completion), patch("backend.ai_agent.execute_kubectl") as execute: - events = await collect_events(run_diagnostic_agent("report", max_steps=3, execution_mode="ai")) - return events, execute + executed = [] - events, execute = asyncio.run(run()) - execute.assert_not_called() + async def run(): + async def fake_execute(command): + executed.append(command) + return {"command": command, "mode": "read", "returncode": 0, "stdout": "nodes ready", "stderr": ""} + + 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", max_steps=3, execution_mode="ai")) + + events = asyncio.run(run()) + self.assertEqual(executed, ["kubectl get nodes -o wide"]) self.assertEqual(events[-1]["type"], "approval_required") def test_rejects_unknown_execution_mode(self): @@ -80,6 +88,39 @@ class DiagnosticAgentTest(unittest.TestCase): self.assertEqual(events[-1]["type"], "error") self.assertIn("执行模式", events[-1]["message"]) + def test_ai_mode_forces_initial_read_only_probe_before_model_planning(self): + responses = iter([ + '{"action":"final","analysis":"根据采集结果完成分析"}', + ]) + + async def fake_completion(messages): + return next(responses) + + executed = [] + + async def fake_execute(command): + executed.append(command) + return {"command": command, "mode": "read", "returncode": 0, "stdout": "node-a Ready", "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", max_steps=3, execution_mode="ai")) + + events = asyncio.run(run()) + self.assertEqual(executed, ["kubectl get nodes -o wide"]) + self.assertEqual([event["type"] for event in events[:2]], ["command", "result"]) + self.assertEqual(events[-1]["type"], "final") + + def test_parse_agent_action_accepts_markdown_json_fence(self): + from backend.ai_agent import _parse_agent_action + + action = _parse_agent_action( + '我将先检查 Pod。\n```json\n{"action":"command","command":"kubectl get pods -A","reason":"查看 Pod"}\n```' + ) + + self.assertEqual(action["action"], "command") + self.assertEqual(action["command"], "kubectl get pods -A") + def test_stops_when_model_returns_invalid_action(self): async def fake_completion(messages): return "not-json" diff --git a/tests/test_history_api.py b/tests/test_history_api.py new file mode 100644 index 0000000..4b5904a --- /dev/null +++ b/tests/test_history_api.py @@ -0,0 +1,43 @@ +import asyncio +import os +import tempfile +import unittest +from unittest.mock import patch + +import backend.main as main +from backend.history_store import HistoryStore + + +class HistoryApiTest(unittest.TestCase): + def setUp(self): + fd, self.path = tempfile.mkstemp(suffix=".db") + os.close(fd) + self.store = HistoryStore(self.path) + self.original_store = main.history_store + main.history_store = self.store + + def tearDown(self): + main.history_store = self.original_store + os.remove(self.path) + + def test_lists_and_gets_full_diagnosis_history(self): + record_id = self.store.save_diagnosis( + {"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 50, "status": "critical"}, + "report", + ) + + records = asyncio.run(main.list_diagnosis_history()) + detail = asyncio.run(main.get_diagnosis_history(record_id)) + + self.assertEqual(records["total"], 1) + self.assertEqual(records["items"][0]["id"], record_id) + self.assertEqual(detail["report"], "report") + + def test_get_history_returns_404_for_unknown_id(self): + with self.assertRaises(main.HTTPException) as context: + asyncio.run(main.get_diagnosis_history("missing")) + self.assertEqual(context.exception.status_code, 404) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_history_store.py b/tests/test_history_store.py new file mode 100644 index 0000000..545953c --- /dev/null +++ b/tests/test_history_store.py @@ -0,0 +1,69 @@ +import os +import tempfile +import unittest + +from backend.history_store import HistoryStore + + +class HistoryStoreTest(unittest.TestCase): + def setUp(self): + fd, self.path = tempfile.mkstemp(suffix=".db") + os.close(fd) + self.store = HistoryStore(self.path) + + def tearDown(self): + os.remove(self.path) + + def test_persists_diagnosis_and_can_query_after_reopen(self): + diagnosis = { + "timestamp": "2026-07-25T12:00:00", + "namespace": "prod", + "score": 65, + "status": "warning", + "critical_count": 1, + "warning_count": 4, + "issues": [{"level": "critical", "message": "pod failed"}], + "checks": {}, + } + record_id = self.store.save_diagnosis(diagnosis, "report text") + + reopened = HistoryStore(self.path) + records = reopened.list_diagnoses() + detail = reopened.get_diagnosis(record_id) + + self.assertEqual(len(records), 1) + self.assertEqual(records[0]["namespace"], "prod") + self.assertEqual(detail["diagnosis"]["score"], 65) + self.assertEqual(detail["report"], "report text") + + def test_records_all_processing_events_in_order(self): + diagnosis_id = self.store.save_diagnosis( + {"timestamp": "2026-07-25T12:00:00", "namespace": "all", "score": 80, "status": "warning"}, + "report", + ) + run_id = self.store.create_run(diagnosis_id, "agent", "ai", "check pods") + self.store.append_event(run_id, "command", {"command": "kubectl get pods -A"}) + self.store.append_event(run_id, "result", {"returncode": 0, "stdout": "ok"}) + self.store.finish_run(run_id, "completed", "done") + + detail = self.store.get_diagnosis(diagnosis_id) + + self.assertEqual(len(detail["runs"]), 1) + self.assertEqual(detail["runs"][0]["status"], "completed") + self.assertEqual([event["type"] for event in detail["runs"][0]["events"]], ["command", "result"]) + + def test_lists_all_diagnoses_newest_first_and_filters_status(self): + first = {"timestamp": "2026-07-25T11:00:00", "namespace": "dev", "score": 100, "status": "healthy"} + second = {"timestamp": "2026-07-25T12:00:00", "namespace": "prod", "score": 20, "status": "critical"} + self.store.save_diagnosis(first, "first") + self.store.save_diagnosis(second, "second") + + all_records = self.store.list_diagnoses() + critical = self.store.list_diagnoses(status="critical") + + self.assertEqual([item["namespace"] for item in all_records], ["prod", "dev"]) + self.assertEqual([item["namespace"] for item in critical], ["prod"]) + + +if __name__ == "__main__": + unittest.main()