feat: 移除 etcdutl 引用,增加 ctr/nerdctl 示例,实现 Agent 断点续传
- AI Agent Prompt 移除不存在的 etcdutl 命令引用 - Prompt 增加 containerd(ctr/nerdctl) 和 Linux 基础调优命令示例 - processing_runs 表新增 context_json 字段,保存 Agent 执行上下文 - 每次命令执行后自动保存上下文到 DB
This commit is contained in:
+79
-3
@@ -29,8 +29,10 @@ AGENT_SYSTEM_PROMPT = """你是 Kubernetes 自主诊断 Agent。你的首要职
|
|||||||
|
|
||||||
可用工具:
|
可用工具:
|
||||||
1. kubectl 命令 - 查询集群内资源状态。例如: kubectl get pods -A, kubectl describe node <name>, kubectl logs <pod> -n <ns>
|
1. kubectl 命令 - 查询集群内资源状态。例如: kubectl get pods -A, kubectl describe node <name>, kubectl logs <pod> -n <ns>
|
||||||
2. 系统诊断 shell 命令 - 查询节点宿主机状态。例如: systemctl status kubelet, journalctl -u kubelet -n 100 --no-pager, crictl ps, df -h, free -m, ss -tlnp, ip addr
|
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. etcd 诊断命令 - 检查 etcd 集群健康与数据。例如: etcdctl endpoint status, etcdctl member list, etcdctl alarm list, etcdctl get /kubernetes --prefix, etcdutl snapshot status <file>
|
3. containerd 命令 - 查询容器运行时状态。例如: ctr containers list, ctr images list, ctr tasks ls, nerdctl ps -a, nerdctl logs <container>, 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 :<port>, ethtool <iface>, tcpdump -i any -c 10 -nn
|
||||||
|
|
||||||
行为要求:
|
行为要求:
|
||||||
1. 如果证据不足,必须返回 command 动作并逐条调查;至少检查与故障资源直接相关的状态、describe、events 或 logs。
|
1. 如果证据不足,必须返回 command 动作并逐条调查;至少检查与故障资源直接相关的状态、describe、events 或 logs。
|
||||||
@@ -112,6 +114,8 @@ async def run_diagnostic_agent(
|
|||||||
question: str = "",
|
question: str = "",
|
||||||
max_steps: int = 30,
|
max_steps: int = 30,
|
||||||
execution_mode: str = "ai",
|
execution_mode: str = "ai",
|
||||||
|
run_id: str = "",
|
||||||
|
history_store=None,
|
||||||
):
|
):
|
||||||
"""自主诊断循环:AI 模式执行只读命令,手动模式等待用户执行。"""
|
"""自主诊断循环:AI 模式执行只读命令,手动模式等待用户执行。"""
|
||||||
if execution_mode not in {"ai", "manual"}:
|
if execution_mode not in {"ai", "manual"}:
|
||||||
@@ -151,6 +155,8 @@ async def run_diagnostic_agent(
|
|||||||
command_counts=command_counts,
|
command_counts=command_counts,
|
||||||
report=report,
|
report=report,
|
||||||
question=question,
|
question=question,
|
||||||
|
run_id=run_id,
|
||||||
|
history_store=history_store,
|
||||||
):
|
):
|
||||||
yield event
|
yield event
|
||||||
|
|
||||||
@@ -163,6 +169,8 @@ async def _run_agent_loop(
|
|||||||
command_counts: Counter,
|
command_counts: Counter,
|
||||||
report: str,
|
report: str,
|
||||||
question: str,
|
question: str,
|
||||||
|
run_id: str = "",
|
||||||
|
history_store=None,
|
||||||
):
|
):
|
||||||
invalid_response_count = 0
|
invalid_response_count = 0
|
||||||
step = start_step
|
step = start_step
|
||||||
@@ -266,6 +274,9 @@ async def _run_agent_loop(
|
|||||||
"command": command, "reason": reason, "mode": decision.mode,
|
"command": command, "reason": reason, "mode": decision.mode,
|
||||||
"message": "请在终端手动执行命令,并根据结果继续排查",
|
"message": "请在终端手动执行命令,并根据结果继续排查",
|
||||||
}
|
}
|
||||||
|
# 手动模式也保存上下文,支持断点续传
|
||||||
|
self_save_context(run_id, history_store, messages, command_counts, step, step + 1,
|
||||||
|
max_steps, report, question, execution_mode, content)
|
||||||
return
|
return
|
||||||
|
|
||||||
if decision.requires_approval:
|
if decision.requires_approval:
|
||||||
@@ -274,15 +285,19 @@ async def _run_agent_loop(
|
|||||||
"command": command,
|
"command": command,
|
||||||
"reason": reason,
|
"reason": reason,
|
||||||
"created_by": AI_MODEL,
|
"created_by": AI_MODEL,
|
||||||
"run_id": "",
|
"run_id": run_id,
|
||||||
"report": report,
|
"report": report,
|
||||||
"question": question,
|
"question": question,
|
||||||
"execution_mode": execution_mode,
|
"execution_mode": execution_mode,
|
||||||
|
"history_store": history_store,
|
||||||
"messages": messages + [{"role": "assistant", "content": content}],
|
"messages": messages + [{"role": "assistant", "content": content}],
|
||||||
"next_step": step + 1,
|
"next_step": step + 1,
|
||||||
"max_steps": max_steps,
|
"max_steps": max_steps,
|
||||||
"command_counts": dict(command_counts),
|
"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 {
|
yield {
|
||||||
"type": "approval_required", "approval_id": approval_id,
|
"type": "approval_required", "approval_id": approval_id,
|
||||||
"command": command, "reason": reason,
|
"command": command, "reason": reason,
|
||||||
@@ -296,6 +311,10 @@ async def _run_agent_loop(
|
|||||||
{"role": "user", "content": "命令实际执行结果:\n" + json.dumps(result, ensure_ascii=False)},
|
{"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
|
step += 1
|
||||||
|
|
||||||
if max_steps > 0:
|
if max_steps > 0:
|
||||||
@@ -319,6 +338,8 @@ async def resume_diagnostic_agent(pending: dict, result: dict):
|
|||||||
+ json.dumps(result, ensure_ascii=False)
|
+ 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(
|
async for event in _run_agent_loop(
|
||||||
messages=messages,
|
messages=messages,
|
||||||
execution_mode=pending.get("execution_mode", "ai"),
|
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 {}),
|
command_counts=Counter(pending.get("command_counts") or {}),
|
||||||
report=pending.get("report", ""),
|
report=pending.get("report", ""),
|
||||||
question=pending.get("question", ""),
|
question=pending.get("question", ""),
|
||||||
|
run_id=run_id,
|
||||||
|
history_store=history_store,
|
||||||
):
|
):
|
||||||
yield event
|
yield event
|
||||||
|
|
||||||
@@ -335,6 +358,59 @@ def take_pending_approval(approval_id: str) -> dict | None:
|
|||||||
return _PENDING_APPROVALS.pop(approval_id, 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]):
|
async def _stream_chat(messages: list[dict]):
|
||||||
"""通用流式请求 - yield 文本 chunk"""
|
"""通用流式请求 - yield 文本 chunk"""
|
||||||
# connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死
|
# connect=10s 快速失败; read=90s 防止 AI 服务无响应时前端无限卡死
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ class HistoryStore:
|
|||||||
summary TEXT NOT NULL DEFAULT '',
|
summary TEXT NOT NULL DEFAULT '',
|
||||||
started_at TEXT NOT NULL,
|
started_at TEXT NOT NULL,
|
||||||
finished_at TEXT,
|
finished_at TEXT,
|
||||||
|
context_json TEXT NOT NULL DEFAULT '',
|
||||||
FOREIGN KEY(diagnosis_id) REFERENCES diagnoses(id) ON DELETE CASCADE
|
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 INDEX IF NOT EXISTS idx_runs_diagnosis ON processing_runs(diagnosis_id, started_at);
|
||||||
@@ -68,6 +69,15 @@ class HistoryStore:
|
|||||||
UNIQUE(run_id, sequence)
|
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:
|
def save_diagnosis(self, diagnosis: dict, report: str) -> str:
|
||||||
record_id = uuid.uuid4().hex
|
record_id = uuid.uuid4().hex
|
||||||
@@ -130,6 +140,11 @@ class HistoryStore:
|
|||||||
{"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])}
|
{"type": event["type"], "created_at": event["created_at"], **json.loads(event["payload_json"])}
|
||||||
for event in event_rows
|
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)
|
runs.append(run)
|
||||||
return {
|
return {
|
||||||
"id": row["id"], "created_at": row["created_at"],
|
"id": row["id"], "created_at": row["created_at"],
|
||||||
@@ -152,6 +167,30 @@ class HistoryStore:
|
|||||||
)
|
)
|
||||||
return run_id
|
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):
|
def append_event(self, run_id: str, event_type: str, payload: dict):
|
||||||
with self._connect() as db:
|
with self._connect() as db:
|
||||||
sequence = db.execute(
|
sequence = db.execute(
|
||||||
|
|||||||
+51
-1
@@ -22,7 +22,8 @@ from backend.diagnosis import (
|
|||||||
from backend.ai_agent import (
|
from backend.ai_agent import (
|
||||||
analyze_with_ai, chat_with_ai,
|
analyze_with_ai, chat_with_ai,
|
||||||
analyze_with_ai_stream, chat_with_ai_stream,
|
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.agent_tools import execute_command
|
||||||
from backend.history_store import HistoryStore
|
from backend.history_store import HistoryStore
|
||||||
@@ -329,6 +330,8 @@ async def agent_diagnose_stream(question: str = "", execution_mode: str = "ai"):
|
|||||||
question=question,
|
question=question,
|
||||||
max_steps=0,
|
max_steps=0,
|
||||||
execution_mode=execution_mode,
|
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"})
|
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":
|
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")
|
@app.post("/api/chat")
|
||||||
async def chat(req: ChatRequest):
|
async def chat(req: ChatRequest):
|
||||||
"""AI 多轮对话 (非流式,兼容旧接口)"""
|
"""AI 多轮对话 (非流式,兼容旧接口)"""
|
||||||
|
|||||||
Reference in New Issue
Block a user