diff --git a/backend/ai_agent.py b/backend/ai_agent.py index e1524f6..83eaa56 100644 --- a/backend/ai_agent.py +++ b/backend/ai_agent.py @@ -29,8 +29,10 @@ AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职 可用工具： 1. kubectl 命令 - 查询集群内资源状态。例如: kubectl get pods -A, kubectl describe node , kubectl logs -n -2. 系统诊断 shell 命令 - 查询节点宿主机状态。例如: systemctl status kubelet, journalctl -u kubelet -n 100 --no-pager, crictl ps, df -h, free -m, ss -tlnp, ip addr -3. etcd 诊断命令 - 检查 etcd 集群健康与数据。例如: etcdctl endpoint status, etcdctl member list, etcdctl alarm list, etcdctl get /kubernetes --prefix, etcdutl snapshot status +2. 系统诊断 shell 命令 - 查询节点宿主机状态。例如: systemctl status kubelet, journalctl -u kubelet -n 100 --no-pager, crictl ps, df -h, free -m, ss -tlnp, ip addr, top, du -sh, curl, tcpdump +3. containerd 命令 - 查询容器运行时状态。例如: ctr containers list, ctr images list, ctr tasks ls, nerdctl ps -a, nerdctl logs , nerdctl stats +4. etcd 诊断命令 - 检查 etcd 集群健康与数据。例如: etcdctl endpoint status, etcdctl endpoint health, etcdctl member list, etcdctl alarm list, etcdctl get /kubernetes --prefix +5. Linux 基础调优命令 - 例如: iostat -xz 1 3, vmstat 1 3, mpstat -P ALL 1 3, pidstat -d 1 3, lsof -i :, ethtool , tcpdump -i any -c 10 -nn 行为要求： 1. 如果证据不足，必须返回 command 动作并逐条调查；至少检查与故障资源直接相关的状态、describe、events 或 logs。 @@ -112,6 +114,8 @@ async def run_diagnostic_agent( question: str = "", max_steps: int = 30, execution_mode: str = "ai", + run_id: str = "", + history_store=None, ): """自主诊断循环：AI 模式执行只读命令，手动模式等待用户执行。""" if execution_mode not in {"ai", "manual"}: @@ -151,6 +155,8 @@ async def run_diagnostic_agent( command_counts=command_counts, report=report, question=question, + run_id=run_id, + history_store=history_store, ): yield event @@ -163,6 +169,8 @@ async def _run_agent_loop( command_counts: Counter, report: str, question: str, + run_id: str = "", + history_store=None, ): invalid_response_count = 0 step = start_step @@ -266,6 +274,9 @@ async def _run_agent_loop( "command": command, "reason": reason, "mode": decision.mode, "message": "请在终端手动执行命令，并根据结果继续排查", } + # 手动模式也保存上下文，支持断点续传 + self_save_context(run_id, history_store, messages, command_counts, step, step + 1, + max_steps, report, question, execution_mode, content) return if decision.requires_approval: @@ -274,15 +285,19 @@ async def _run_agent_loop( "command": command, "reason": reason, "created_by": AI_MODEL, - "run_id": "", + "run_id": run_id, "report": report, "question": question, "execution_mode": execution_mode, + "history_store": history_store, "messages": messages + [{"role": "assistant", "content": content}], "next_step": step + 1, "max_steps": max_steps, "command_counts": dict(command_counts), } + # 保存上下文到 DB，避免浏览器关闭/刷新丢失进度 + self_save_context(run_id, history_store, messages, command_counts, step, step + 1, + max_steps, report, question, execution_mode, content) yield { "type": "approval_required", "approval_id": approval_id, "command": command, "reason": reason, @@ -296,6 +311,10 @@ async def _run_agent_loop( {"role": "user", "content": "命令实际执行结果：\n" + json.dumps(result, ensure_ascii=False)}, ]) + # 每次执行后保存上下文，支持断点续传 + self_save_context(run_id, history_store, messages, command_counts, step, step + 1, + max_steps, report, question, execution_mode) + step += 1 if max_steps > 0: @@ -319,6 +338,8 @@ async def resume_diagnostic_agent(pending: dict, result: dict): + json.dumps(result, ensure_ascii=False) ), }) + run_id = pending.get("run_id", "") + history_store = pending.get("history_store") async for event in _run_agent_loop( messages=messages, execution_mode=pending.get("execution_mode", "ai"), @@ -327,6 +348,8 @@ async def resume_diagnostic_agent(pending: dict, result: dict): command_counts=Counter(pending.get("command_counts") or {}), report=pending.get("report", ""), question=pending.get("question", ""), + run_id=run_id, + history_store=history_store, ): yield event @@ -335,6 +358,59 @@ def take_pending_approval(approval_id: str) -> dict | None: return _PENDING_APPROVALS.pop(approval_id, None) +def self_save_context( + run_id: str, history_store, + messages: list, command_counts: Counter, + step: int, next_step: int, + max_steps: int, report: str, question: str, + execution_mode: str, + last_response_content: str = "", +): + """保存 Agent 执行上下文到 DB，支持断点续传。""" + if not run_id or not history_store: + return + context = { + "messages": messages, + "command_counts": dict(command_counts), + "step": step, + "next_step": next_step, + "max_steps": max_steps, + "report": report, + "question": question, + "execution_mode": execution_mode, + "last_response": last_response_content, + } + history_store.update_context(run_id, context) + + +async def resume_from_context( + ctx: dict, + run_id: str = "", + history_store=None, +): + """从 DB 保存的上下文恢复 Agent 诊断，不重跑历史命令，直接进入下一轮。""" + messages = list(ctx.get("messages") or []) + command_counts = Counter(ctx.get("command_counts") or {}) + next_step = int(ctx.get("next_step", 1)) + max_steps = int(ctx.get("max_steps", 0)) + report = ctx.get("report", "") + question = ctx.get("question", "") + execution_mode = ctx.get("execution_mode", "ai") + + async for event in _run_agent_loop( + messages=messages, + execution_mode=execution_mode, + max_steps=max_steps, + start_step=next_step, + command_counts=command_counts, + report=report, + question=question, + run_id=run_id, + history_store=history_store, + ): + yield event + + async def _stream_chat(messages: list[dict]): """通用流式请求 - yield 文本 chunk""" # connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死 diff --git a/backend/history_store.py b/backend/history_store.py index ca6e7b0..07356f3 100644 --- a/backend/history_store.py +++ b/backend/history_store.py @@ -53,6 +53,7 @@ class HistoryStore: summary TEXT NOT NULL DEFAULT '', started_at TEXT NOT NULL, finished_at TEXT, + context_json TEXT NOT NULL DEFAULT '', 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); @@ -68,6 +69,15 @@ class HistoryStore: UNIQUE(run_id, sequence) ); """) + # 兼容旧表：确保 context_json 列存在 + self._ensure_column(db, "processing_runs", "context_json", "TEXT NOT NULL DEFAULT ''") + + def _ensure_column(self, db, table: str, column: str, col_def: str): + try: + db.execute(f"ALTER TABLE {table} ADD COLUMN {column} {col_def}") + except sqlite3.OperationalError as e: + if "duplicate column" not in str(e).lower(): + raise def save_diagnosis(self, diagnosis: dict, report: str) -> str: record_id = uuid.uuid4().hex @@ -130,6 +140,11 @@ class HistoryStore: {"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])} for event in event_rows ] + run["context_json"] = run.get("context_json") or "" + if run["context_json"]: + run["context"] = json.loads(run["context_json"]) + else: + run["context"] = None runs.append(run) return { "id": row["id"], "created_at": row["created_at"], @@ -152,6 +167,30 @@ class HistoryStore: ) return run_id + def update_context(self, run_id: str, context: dict): + """保存/更新 Agent 执行上下文到 processing_runs，支持断点续传。""" + with self._connect() as db: + db.execute( + "UPDATE processing_runs SET context_json = ? WHERE id = ?", + (json.dumps(context, ensure_ascii=False), run_id), + ) + + def get_context(self, run_id: str) -> Optional[dict]: + """读取 Agent 执行上下文，如果不存在或为空则返回 None。""" + with self._connect() as db: + row = db.execute( + "SELECT context_json, diagnosis_id, report FROM processing_runs pr JOIN diagnoses d ON pr.diagnosis_id = d.id WHERE pr.id = ?", + (run_id,), + ).fetchone() + if row is None: + return None + ctx_str = row["context_json"] or "" + if not ctx_str: + return None + ctx = json.loads(ctx_str) + ctx["report"] = row["report"] + return ctx + def append_event(self, run_id: str, event_type: str, payload: dict): with self._connect() as db: sequence = db.execute( diff --git a/backend/main.py b/backend/main.py index 0b3cf00..5da8b14 100644 --- a/backend/main.py +++ b/backend/main.py @@ -22,7 +22,8 @@ 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, resume_diagnostic_agent, take_pending_approval, + run_diagnostic_agent, resume_diagnostic_agent, resume_from_context, + take_pending_approval, ) from backend.agent_tools import execute_command from backend.history_store import HistoryStore @@ -329,6 +330,8 @@ async def agent_diagnose_stream(question: str = "", execution_mode: str = "ai"): question=question, max_steps=0, execution_mode=execution_mode, + run_id=run_id, + history_store=history_store, ): 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": @@ -409,6 +412,53 @@ async def execute_approved_agent_command(req: AgentApprovalRequest): ) +@app.get("/api/agent/resume-stream/{run_id}") +async def agent_resume_stream(run_id: str): + """从 DB 上下文恢复 Agent 诊断，不重跑历史命令，直接继续诊断。""" + ctx = history_store.get_context(run_id) + if ctx is None: + raise HTTPException(status_code=404, detail="该运行记录不存在或不支持断点续传") + ctx.pop("report", None) # get_context 已经注入了 report + + async def event_generator(): + yield f"data: {json.dumps({'type': 'start', 'mode': ctx.get('execution_mode', 'ai'), 'run_id': run_id, 'resumed': True}, ensure_ascii=False)}\n\n" + final_status = "completed" + summary = "" + try: + async for event in resume_from_context( + ctx=ctx, + run_id=run_id, + history_store=history_store, + ): + 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" + 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: + 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(), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, + ) + + @app.post("/api/chat") async def chat(req: ChatRequest): """AI 多轮对话 (非流式，兼容旧接口)"""