一、应用启动流程
1.1 入口点
文件: app/main.py
───────────────────────────────────────────────────────────────────
行 86-93: run() 函数（应用启动入口）
───────────────────────────────────────────────────────────────────
86→ def run() -> None:
87→ import uvicorn
88→
89→ uvicorn.run("app.main:app", host=settings.app_host, port=settings.app_port, reload=True)
90→
91→
92→ if __name__ == "__main__":
93→ run()
───────────────────────────────────────────────────────────────────1.2 FastAPI 应用实例化与生命周期
文件: app/main.py
───────────────────────────────────────────────────────────────────
行 17-46: lifespan 上下文管理器
───────────────────────────────────────────────────────────────────
17→ @asynccontextmanager
18→ async def lifespan(app: FastAPI):
19→ """应用生命周期管理"""
20→ init_db() # 行 20: 初始化数据库
21→
22→ # 👇 启动时验证模型配置
23→ try:
24→ from app.agent.runtime import _get_default_model # 行 24
25→ from app.llm.registry import build_chat_model # 行 25
26→
27→ model_cfg = _get_default_model() # 行 27: 获取默认模型
28→ ...
34→ model = build_chat_model(model_cfg) # 行 37: 创建模型实例
35→ ...
49→ app = FastAPI(title="运维 Agent", version="0.1.0", lifespan=lifespan) # 行 49
───────────────────────────────────────────────────────────────────
行 55-64: 认证中间件
───────────────────────────────────────────────────────────────────
55→ @app.middleware("http")
56→ async def _auth_gate(request: Request, call_next):
57→ if settings.auth_enabled:
58→ p = request.url.path
59→ if not (p in _OPEN_EXACT or p.startswith("/auth") or p.startswith("/static")):
60→ data = verify_token(request.cookies.get(auth.COOKIE_NAME), ...)
61→ if not (data and data.get("u") == settings.auth_username):
62→ return JSONResponse({"detail": "未认证,请先登录"}, status_code=401)
63→ return await call_next(request)
───────────────────────────────────────────────────────────────────
行 71-74: 路由注册
───────────────────────────────────────────────────────────────────
71→ app.include_router(auth.router) # 行 71
72→ app.include_router(admin.router) # 行 72
73→ app.include_router(chat.router) # 行 73
74→ app.include_router(terminal.router) # 行 74
───────────────────────────────────────────────────────────────────
行 81-83: 首页路由
───────────────────────────────────────────────────────────────────
81→ @app.get("/")
82→ def index() -> FileResponse:
83→ return FileResponse(_web_dir / "index.html")
───────────────────────────────────────────────────────────────────1.3 数据库初始化
文件: app/db/base.py
───────────────────────────────────────────────────────────────────
行 28-32: init_db() 函数
───────────────────────────────────────────────────────────────────
28→ def init_db() -> None:
29→ from app.db import models # noqa: F401 注册所有表
30→
31→ Base.metadata.create_all(bind=engine) # 行 31: 创建表
32→ _migrate_add_columns() # 行 32: 轻量迁移
───────────────────────────────────────────────────────────────────
行 35-46: 轻量迁移函数
───────────────────────────────────────────────────────────────────
35→ def _migrate_add_columns() -> None:
36→ """轻量迁移:为已存在的表补加新增列"""
37→ wanted = {"servers": [("ssh_key_id", "INTEGER")]} # 行 37
38→ insp = inspect(engine)
39→ with engine.begin() as conn:
40→ for table, cols in wanted.items():
41→ if not insp.has_table(table):
42→ continue
43→ existing = {c["name"] for c in insp.get_columns(table)}
44→ for col, coltype in cols:
45→ if col not in existing:
46→ conn.execute(text(f"ALTER TABLE {table} ADD COLUMN {col} {coltype}"))
───────────────────────────────────────────────────────────────────二、用户登录认证流程
文件: app/api/auth.py
───────────────────────────────────────────────────────────────────
行 26-37: login() 登录函数
───────────────────────────────────────────────────────────────────
26→ @router.post("/login")
27→ def login(body: LoginIn, response: Response):
28→ if not settings.auth_enabled:
29→ return {"ok": True, "auth_enabled": False}
30→ ok = (secrets.compare_digest(body.username, settings.auth_username) # 行 30
31→ and secrets.compare_digest(body.password, settings.auth_password)) # 行 31
32→ if not ok:
33→ raise HTTPException(401, "用户名或密码错误")
34→ token = sign_token({"u": body.username, "t": int(time.time())}) # 行 34
35→ response.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax",
36→ max_age=settings.session_max_age, path="/")
37→ return {"ok": True}
───────────────────────────────────────────────────────────────────
行 40-43: logout() 登出函数
───────────────────────────────────────────────────────────────────
40→ @router.post("/logout")
41→ def logout(response: Response):
42→ response.delete_cookie(COOKIE_NAME, path="/")
43→ return {"ok": True}
───────────────────────────────────────────────────────────────────
文件: app/db/crypto.py
───────────────────────────────────────────────────────────────────
行 32-43: 令牌签名与验证
───────────────────────────────────────────────────────────────────
32→ def sign_token(payload: dict) -> str:
33→ return _fernet().encrypt(json.dumps(payload).encode()).decode() # 行 33
34→
35→
36→ def verify_token(token: str | None, max_age: int | None = None) -> dict | None:
37→ if not token:
38→ return None
39→ try:
40→ raw = _fernet().decrypt(token.encode(), ttl=max_age) # 行 40: 解密+TTL验证
41→ return json.loads(raw)
42→ except (InvalidToken, ValueError):
43→ return None
───────────────────────────────────────────────────────────────────三、Agent 对话核心流程
3.1 对话入口 API
文件: app/api/chat.py
───────────────────────────────────────────────────────────────────
行 52-62: chat_stream() 对话发起
───────────────────────────────────────────────────────────────────
52→ @router.post("/stream")
53→ async def chat_stream(body: ChatIn):
54→ thread_id = body.thread_id or uuid.uuid4().hex # 行 54
55→
56→ async def gen():
57→ yield {"type": "thread", "thread_id": thread_id}
58→ async for ev in astream_turn(thread_id, body.message,
59→ body.servers, body.clouds,
60→ body.auto_approve_all): # 行 58-60
61→ yield ev
62→ return _sse(gen())
行 44-49: SSE 流式响应辅助函数
───────────────────────────────────────────────────────────────────
44→ def _sse(gen: AsyncIterator[dict]) -> StreamingResponse:
45→ async def event_stream():
46→ async for ev in gen:
47→ yield f"data: {json.dumps(ev, ensure_ascii=False)}\n\n"
48→ return StreamingResponse(event_stream(), media_type="text/event-stream", ...)
───────────────────────────────────────────────────────────────────3.2 Agent 运行时核心
文件: app/agent/runtime.py
───────────────────────────────────────────────────────────────────
行 177-189: astream_turn() 主入口
───────────────────────────────────────────────────────────────────
177→ async def astream_turn(thread_id: str, user_message: str,
178→ servers: list[str] | None = None,
179→ clouds: list[str] | None = None,
180→ auto_approve_all: bool = False) -> AsyncIterator[dict]:
181→ _ensure_conversation(thread_id, title=user_message) # 行 181: 创建会话记录
182→ _save_message(thread_id, "user", user_message) # 行 182: 保存用户消息
183→ agent = await _assemble(servers, clouds) # 行 183: 组装 Agent
184→ config = {"configurable": {"thread_id": thread_id}} # 行 184
185→ graph_input = {"messages": [HumanMessage(content=user_message)], "plan": [],
186→ "current_step": 0, "dry_run": settings.default_dry_run,
187→ "notes": "", "approved_ids": [],
188→ "auto_approve_all": auto_approve_all, "last_io": {}}
189→ async for ev in _run_stream(agent, graph_input, config, thread_id): # 行 189
190→ yield ev
───────────────────────────────────────────────────────────────────
行 51-69: _assemble() 组装 Agent
───────────────────────────────────────────────────────────────────
51→ async def _assemble(servers: list[str] | None = None,
52→ clouds: list[str] | None = None):
53→ model_cfg = _get_default_model() # 行 53: 获取默认模型
54→ model = build_chat_model(model_cfg) # 行 54: 构建模型实例
55→
56→ allowed = set(servers) if servers else None # 行 56
57→ tools = [*make_scoped_ssh_tools(allowed), clarify_tool] # 行 57
58→
59→ with SessionLocal() as db:
60→ q = db.query(CloudAccount).filter(CloudAccount.enabled)
61→ if clouds:
62→ q = q.filter(CloudAccount.name.in_(clouds))
63→ accounts = q.all()
64→ for a in accounts:
65→ db.expunge(a)
66→ tools += await load_cloud_tools(accounts) # 行 66: 加载 MCP 云工具
67→
68→ suffix = _scope_suffix(servers, clouds) # 行 68
69→ return build_agent(model, tools, checkpointer=_checkpointer, system_suffix=suffix) # 行 69
───────────────────────────────────────────────────────────────────
行 41-48: _get_default_model() 获取默认模型
───────────────────────────────────────────────────────────────────
41→ def _get_default_model() -> ModelProvider:
42→ with SessionLocal() as db:
43→ cfg = (db.query(ModelProvider).filter(ModelProvider.is_default).first() # 行 43
44→ or db.query(ModelProvider).first()) # 行 44
45→ if cfg is None:
46→ raise RuntimeError("尚未配置任何模型供应商,请先在后台添加。")
47→ db.expunge(cfg)
48→ return cfg
───────────────────────────────────────────────────────────────────
行 112-175: _run_stream() 流式事件处理
───────────────────────────────────────────────────────────────────
112→ async def _run_stream(agent, graph_input, config, thread_id: str) -> AsyncIterator[dict]:
113→ assistant_buf: list[str] = []
114→ finished = False
115→ try:
116→ async for mode, payload in agent.astream( # 行 116
117→ graph_input, config, stream_mode=["updates", "messages"]
118→ ):
119→ if mode == "messages": # 行 119
120→ chunk, meta = payload
121→ text = chunk.content if isinstance(chunk.content, str) else ""
122→ if text and meta.get("langgraph_node") == "agent":
123→ assistant_buf.append(text)
124→ yield {"type": "token", "text": text} # 行 124
125→ elif mode == "updates": # 行 126
126→ if "__interrupt__" in payload: # 行 127
127→ if assistant_buf:
128→ _save_message(thread_id, "assistant", "".join(assistant_buf))
129→ assistant_buf = []
130→ _set_status(thread_id, "waiting_approval") # 行 131
131→ finished = True
132→ intr = dict(payload["__interrupt__"][0].value) # 行 133
133→ intr.setdefault("type", "approval_required")
134→ yield intr # 行 135
135→ return # 行 136
138→ for node, upd in payload.items(): # 行 138
139→ if node == "agent": # 行 139
140→ if (upd or {}).get("last_io"):
141→ yield {"type": "llm_io", **upd["last_io"]}
142→ msgs = (upd or {}).get("messages") or []
143→ if msgs and getattr(msgs[-1], "tool_calls", None):
144→ for tc in msgs[-1].tool_calls:
145→ level, _ = classify_tool_call(tc["name"], tc["args"]) # 行 145
146→ _save_message(thread_id, "tool_call", ...) # 行 147
147→ yield {"type": "tool_call", ...} # 行 151
148→ elif node == "execute_tools": # 行 153
149→ for m in (upd or {}).get("messages") or []:
150→ _save_message(thread_id, "tool", m.content, ...)
151→ yield {"type": "tool_result", ...} # 行 156
159→ final = "".join(assistant_buf)
160→ if final:
161→ _save_message(thread_id, "assistant", final)
162→ _set_status(thread_id, "done") # 行 163
163→ finished = True
164→ yield {"type": "done", "reply": final} # 行 165
165→ except Exception as e:
166→ finished = True
167→ yield {"type": "error", "error": f"{type(e).__name__}: {e}"} # 行 168
───────────────────────────────────────────────────────────────────
行 192-203: astream_resume() 审批续跑
───────────────────────────────────────────────────────────────────
192→ async def astream_resume(thread_id: str, action: str, ids: list[str],
193→ remember: bool = False,
194→ auto_approve_all: bool = False,
195→ servers: list[str] | None = None,
196→ clouds: list[str] | None = None) -> AsyncIterator[dict]:
197→ agent = await _assemble(servers, clouds) # 行 197
198→ config = {"configurable": {"thread_id": thread_id}} # 行 198
199→ cmd = Command(resume={"action": action, "ids": ids or [], "remember": remember},
200→ update={"auto_approve_all": auto_approve_all}) # 行 200-201
201→ async for ev in _run_stream(agent, cmd, config, thread_id): # 行 202
202→ yield ev
───────────────────────────────────────────────────────────────────3.3 LangGraph 图构建
文件: app/agent/graph.py
───────────────────────────────────────────────────────────────────
行 38-229: build_agent() 构建 LangGraph
───────────────────────────────────────────────────────────────────
38→ def build_agent(model: BaseChatModel, tools: list[BaseTool],
39→ checkpointer=None, system_suffix: str = ""):
40→ tools_by_name = {t.name: t for t in tools} # 行 40
41→ model_with_tools = model.bind_tools(tools) # 行 41
42→ system_text = SYSTEM_PROMPT + (("\n\n" + system_suffix) if system_suffix else "") # 行 42
44→ # -- 节点:推理 --
45→ async def agent_node(state: AgentState) -> dict: # 行 45
46→ messages = state["messages"]
47→ if not messages or not isinstance(messages[0], SystemMessage):
48→ messages = [SystemMessage(content=system_text), *messages]
49→
50→ # 🔥 真正调用 LLM（添加超时保护）
51→ try:
52→ import asyncio
53→
54→ # 设置超时时间（60秒）
55→ response = await asyncio.wait_for( # 行 55
56→ model_with_tools.ainvoke(messages),
57→ timeout=60.0 # 行 57
58→ )
59→
60→ except asyncio.TimeoutError: # 行 60
61→ error_msg = "LLM API 请求超时（60秒）..."
62→ return {"messages": [AIMessage(content=error_msg)], ...}
63→
64→ except Exception as e: # 行 75
65→ if "402" in error_detail or "insufficient_balance" in error_detail.lower():
66→ error_msg = "❌ 账户余额不足..."
67→ elif "401" in error_detail or "authentication" in error_detail.lower():
68→ error_msg = "❌ API Key 无效..."
69→ elif "404" in error_detail or "not found" in error_detail.lower():
70→ error_msg = "❌ 模型或端点不存在..."
71→ else:
72→ error_msg = f"❌ LLM API 请求失败 ({error_type})..."
97→ usage = getattr(response, "usage_metadata", None) or {}
98→ result = {
99→ "messages": [response],
100→ "last_io": {
101→ "prompt": _render_prompt(messages),
102→ "response": _render_response(response),
103→ "usage": {...}
104→ }
105→ }
106→ return result # 行 106
107→
109→ # -- 节点:护栏 + 逐条审批 --
110→ async def guardrail_node(state: AgentState) -> dict: # 行 115
111→ last: AIMessage = state["messages"][-1]
112→ if state.get("auto_approve_all"): # 行 118
113→ return {"approved_ids": [tc["id"] for tc in last.tool_calls]}
114→ pending = []
115→ auto_ids = []
116→ for tc in last.tool_calls: # 行 122
117→ if not needs_approval(tc["name"]): # 行 123
118→ auto_ids.append(tc["id"])
119→ continue
120→ if tc["name"] == "ssh_run": # 行 126
121→ cmd = tc["args"].get("command", "")
122→ if is_auto_approved(cmd): # 行 128
123→ auto_ids.append(tc["id"])
124→ continue
125→ level = await classify_command_llm(cmd, model) # 行 132
126→ summary = f"在服务器 [{tc['args'].get('server_name')}] 执行:{cmd}"
127→ else:
128→ level, summary = classify_tool_call(tc["name"], tc["args"]) # 行 135
129→ if level == CommandLevel.readonly: # 行 136
130→ auto_ids.append(tc["id"])
131→ else:
132→ pending.append({...}) # 行 139
133→ if not pending or not settings.require_command_approval: # 行 143
134→ return {"approved_ids": [tc["id"] for tc in last.tool_calls]}
136→ decision = interrupt({"type": "approval_required", # 行 147
137→ "operations": pending,
138→ "message": "以下命令需要确认后才会执行"}) or {}
139→ action = decision.get("action", "reject")
140→ if action == "all": # 行 150
141→ approved = [op["id"] for op in pending]
142→ elif action == "selected": # 行 152
143→ approved = [i for i in decision.get("ids", []) if i in {...}]
144→ else:
145→ approved = []
146→ if decision.get("remember"): # 行 157
147→ ...
148→ return {"approved_ids": auto_ids + approved} # 行 162
164→ # -- 节点:执行工具 + 审计 --
165→ async def execute_tools_node(state: AgentState) -> dict: # 行 165
166→ last: AIMessage = state["messages"][-1]
167→ approved = set(state.get("approved_ids") or [])
168→ results = []
169→ for tc in last.tool_calls: # 行 169
170→ level, summary = classify_tool_call(tc["name"], tc["args"])
171→ if tc["id"] not in approved: # 行 171
172→ results.append(ToolMessage(content="[已跳过]...", ...)) # 行 172
173→ continue
174→ tool = tools_by_name.get(tc["name"]) # 行 175
175→ if tool is None:
176→ output, success = f"[错误] 未知工具 {tc['name']}", False
177→ else:
178→ try:
179→ output = str(await tool.ainvoke(tc["args"])) # 行 180
180→ success = not output.startswith("[错误]")
181→ except Exception as e:
182→ output, success = f"[错误] 工具执行异常:{e}", False
183→ results.append(ToolMessage(content=output, ...)) # 行 184
184→ _audit(tc, level, summary, success, output) # 行 185
185→ return {"messages": results, "approved_ids": []} # 行 186
188→ # -- 节点:澄清 --
189→ async def clarify_node(state: AgentState) -> dict: # 行 189
190→ last: AIMessage = state["messages"][-1]
191→ results = []
192→ for tc in last.tool_calls:
193→ if tc["name"] != CLARIFY_TOOL_NAME:
194→ results.append(ToolMessage(content="[已跳过]...", ...))
195→ continue
196→ decision = interrupt({ # 行 198
197→ "type": "clarify_required",
198→ "tool_call_id": tc["id"],
199→ "question": tc["args"].get("question", ""),
200→ "options": tc["args"].get("options", []) or [],
201→ }) or {}
202→ answer = (decision.get("answer") or "").strip()
203→ results.append(ToolMessage(content=(f"用户的选择/补充:{answer}"...), ...))
204→ return {"messages": results}
209→ # -- 路由 --
210→ def route_after_agent(state: AgentState) -> str: # 行 210
211→ last = state["messages"][-1]
212→ if not getattr(last, "tool_calls", None):
213→ return END
214→ if any(tc["name"] == CLARIFY_TOOL_NAME for tc in last.tool_calls):
215→ return "clarify"
216→ return "guardrail"
218→ graph = StateGraph(AgentState) # 行 218
219→ graph.add_node("agent", agent_node) # 行 219
220→ graph.add_node("guardrail", guardrail_node) # 行 220
221→ graph.add_node("execute_tools", execute_tools_node) # 行 221
222→ graph.add_node("clarify", clarify_node) # 行 222
223→ graph.add_edge(START, "agent") # 行 223
224→ graph.add_conditional_edges("agent", route_after_agent, ["guardrail", "clarify", END]) # 行 224
225→ graph.add_edge("guardrail", "execute_tools") # 行 225
226→ graph.add_edge("execute_tools", "agent") # 行 226
227→ graph.add_edge("clarify", "agent") # 行 227
229→ return graph.compile(checkpointer=checkpointer or MemorySaver()) # 行 229
───────────────────────────────────────────────────────────────────
行 252-262: _audit() 审计日志写入
───────────────────────────────────────────────────────────────────
252→ def _audit(tool_call: dict, level: CommandLevel, summary: str,
253→ success: bool, output: str) -> None:
254→ args = tool_call.get("args", {})
255→ target = args.get("server_name") or tool_call["name"].split("__")[0]
256→ command = args.get("command") or summary
257→ with SessionLocal() as db:
258→ db.add(AuditLog(
259→ tool_name=tool_call["name"], target=target, command=command, level=level,
260→ approved=None if level != CommandLevel.dangerous else True,
261→ success=success, output=output[:5000],
262→ ))
263→ db.commit()
───────────────────────────────────────────────────────────────────四、命令分级护栏
文件: app/agent/guardrails.py
───────────────────────────────────────────────────────────────────
行 19-30: 危险命令正则模式
───────────────────────────────────────────────────────────────────
19→ _DANGEROUS = [
20→ r"\brm\s+-[a-z]*[rf]", # rm -rf # 行 20
21→ r"\bmkfs\b", r"\bdd\b", # 行 21
22→ r"\b(shutdown|reboot|halt|poweroff)\b", # 行 22
23→ r"\b(drop|truncate)\s+(table|database)\b", # 行 23
24→ r":\s*\(\)\s*\{", # fork 炸弹 # 行 24
25→ r"\b>\s*/dev/sd", # 直接写磁盘 # 行 25
26→ r"\bchmod\s+-R\s+777\s+/", # 行 26
27→ r"\biptables\s+-F\b", r"\bufw\s+disable\b", # 行 27
28→ r"\buserdel\b", r"\bkill\s+-9\s+1\b", # 行 28
29→ r"\bgit\s+push\s+.*--force", # 行 29
30→ ]
行 33-41: 变更命令正则模式
───────────────────────────────────────────────────────────────────
33→ _MUTATING = [
34→ r"\b(systemctl|service)\s+(restart|stop|start|reload)\b", # 行 34
35→ r"\b(apt|apt-get|yum|dnf|brew)\s+(install|remove|upgrade|update)\b", # 行 35
36→ r"\b(docker|kubectl)\s+(run|rm|delete|apply|restart|scale)\b", # 行 36
37→ r"\b(cp|mv|chmod|chown|ln|mkdir|touch|tee)\b", # 行 37
38→ r"\b(pip|npm|yarn)\s+install\b", # 行 38
39→ r"\b>\s*/", r"\bsed\s+-i\b", # 行 39
40→ r"\bcrontab\b", r"\biptables\b", # 行 40
41→ ]
行 49-57: classify_command() 规则判断
───────────────────────────────────────────────────────────────────
49→ def classify_command(command: str) -> CommandLevel:
50→ cmd = command.lower()
51→ for pat in _DANGEROUS: # 行 51
52→ if re.search(pat, cmd):
53→ return CommandLevel.dangerous # 行 53
54→ for pat in _MUTATING: # 行 54
55→ if re.search(pat, cmd):
56→ return CommandLevel.mutating # 行 56
57→ return CommandLevel.readonly # 行 57
行 70-83: classify_command_llm() LLM 辅助判断
───────────────────────────────────────────────────────────────────
70→ async def classify_command_llm(command: str, model) -> CommandLevel:
71→ if not command.strip():
72→ return CommandLevel.readonly
73→ try:
74→ resp = await model.ainvoke([_LLM_CLASSIFY_SYSTEM, HumanMessage(content=command)]) # 行 75
75→ text = (resp.content if isinstance(resp.content, str) else str(resp.content)).lower()
76→ for level in (CommandLevel.dangerous, CommandLevel.mutating, CommandLevel.readonly): # 行 78
77→ if level.value in text:
78→ return level # 行 79
79→ except Exception:
80→ pass # 行 81
81→ return classify_command(command) # 行 83
行 115-120: needs_approval() 判断是否需要审批
───────────────────────────────────────────────────────────────────
115→ def needs_approval(tool_name: str) -> bool:
116→ """是否属于"会在服务器/云上真正执行"的命令,需要用户确认"""
117→ return tool_name == "ssh_run" or "__" in tool_name # 行 120
───────────────────────────────────────────────────────────────────五、SSH 命令执行
文件: app/tools/ssh.py
───────────────────────────────────────────────────────────────────
行 58-76: _run_on_server() SSH 执行
───────────────────────────────────────────────────────────────────
58→ async def _run_on_server(server_name: str, command: str,
59→ intent: str = "", timeout: int = 600) -> str:
60→ conn_kwargs, err = build_conn_kwargs(server_name) # 行 60
61→ if err:
62→ return f"[错误] {err}"
63→ try:
64→ async with asyncssh.connect(**conn_kwargs) as conn: # 行 63
65→ result = await asyncio.wait_for( # 行 64
66→ conn.run(command, check=False),
67→ timeout=timeout
68→ )
69→ out = (result.stdout or "").strip()
70→ err = (result.stderr or "").strip()
71→ parts = [f"[exit={result.exit_status}]"] # 行 67
72→ if out:
73→ parts.append(f"stdout:\n{out}")
74→ if err:
75→ parts.append(f"stderr:\n{err}")
76→ return "\n".join(parts)
77→ except asyncio.TimeoutError:
78→ return f"[错误] 命令超时(>{timeout}s):{command}" # 行 78
79→ except (OSError, asyncssh.Error) as e:
80→ return f"[错误] SSH 连接/执行失败:{e}" # 行 80
行 26-55: build_conn_kwargs() 构建连接参数
───────────────────────────────────────────────────────────────────
26→ def build_conn_kwargs(server_name: str) -> tuple[dict | None, str | None]:
27→ with SessionLocal() as db:
28→ server = db.query(Server).filter(Server.name == server_name).first() # 行 32
29→ if server is None:
30→ return None, f"未找到服务器 '{server_name}',请先在后台登记。" # 行 34
31→ conn_kwargs: dict = {
32→ "host": server.host,
33→ "port": server.port,
34→ "username": server.username,
35→ "known_hosts": None, # ⚠️ 行 39: 生产应配置 known_hosts
36→ }
37→ if server.auth_type == "key": # 行 41
38→ key_enc, pass_enc = server.private_key_enc, server.passphrase_enc
39→ if server.ssh_key_id is not None: # 行 44
40→ sk = db.get(SSHKey, server.ssh_key_id)
41→ if sk is None:
42→ return None, f"服务器 '{server_name}' 引用的密钥不存在。" # 行 47
43→ key_enc, pass_enc = sk.private_key_enc, sk.passphrase_enc
44→ try:
45→ conn_kwargs["client_keys"] = [asyncssh.import_private_key( # 行 50
46→ decrypt(key_enc), decrypt(pass_enc)
47→ )]
48→ except Exception as e:
49→ return None, f"私钥解析失败:{e}" # 行 52
50→ else:
51→ conn_kwargs["password"] = decrypt(server.password_enc) # 行 54
52→ return conn_kwargs, None # 行 55
行 79-87: ssh_run_tool 定义
───────────────────────────────────────────────────────────────────
79→ ssh_run_tool = StructuredTool.from_function(
80→ coroutine=_run_on_server, # 行 80
81→ name="ssh_run", # 行 81
82→ description="在指定服务器上通过 SSH 执行 shell 命令...", # 行 82-84
83→ args_schema=SSHRunInput, # 行 86
84→ )
行 113-136: make_scoped_ssh_tools() 作用域工具
───────────────────────────────────────────────────────────────────
113→ def make_scoped_ssh_tools(allowed: set[str] | None) -> list[StructuredTool]:
116→ async def _scoped_run(server_name: str, command: str, ...) -> str:
117→ if allowed is not None and server_name not in allowed: # 行 117
118→ return f"[错误] 本次会话被限定只能操作:{', '.join(sorted(allowed))}..."
119→ return await _run_on_server(server_name, command, intent, timeout) # 行 120
122→ def _scoped_list() -> str:
123→ with SessionLocal() as db:
124→ servers = [s for s in db.query(Server).all()
125→ if allowed is None or s.name in allowed] # 行 125
126→ ...
127→ return "可用服务器:\n" + "\n".join(lines) # 行 129
131→ return [
132→ StructuredTool.from_function(func=_scoped_list, name="list_servers", ...), # 行 132
133→ StructuredTool.from_function(coroutine=_scoped_run, name="ssh_run", ...), # 行 134
134→ ]
───────────────────────────────────────────────────────────────────六、交互式终端
文件: app/api/terminal.py
───────────────────────────────────────────────────────────────────
行 37-113: TermSession 类
───────────────────────────────────────────────────────────────────
37→ class TermSession:
38→ def __init__(self, sid: str, server: str, conn, proc) -> None:
39→ self.id = sid
40→ self.server = server
41→ self.conn = conn
42→ self.proc = proc
43→ self.buffer = bytearray()
44→ self.attached: WebSocket | None = None
45→ self.last_active = time.time()
46→ self.closed = False
47→ self.lock = asyncio.Lock()
48→ self.reader = asyncio.create_task(self._read_loop()) # 行 48
50→ async def _read_loop(self) -> None: # 行 50
51→ try:
52→ while True:
53→ data = await self.proc.stdout.read(4096) # 行 54
54→ if not data:
55→ break
56→ async with self.lock:
57→ self.buffer.extend(data) # 行 58
58→ if len(self.buffer) > _BUF_CAP:
59→ del self.buffer[:len(self.buffer) - _BUF_CAP] # 行 60
60→ if self.attached is not None:
61→ try:
62→ await self.attached.send_bytes(data) # 行 63
63→ except Exception:
64→ pass
65→ except Exception:
66→ pass
67→ finally:
68→ await self._on_shell_exit() # 行 69
79→ async def attach(self, ws: WebSocket) -> None: # 行 79
80→ async with self.lock:
81→ if self.attached is not None and self.attached is not ws:
82→ try:
83→ await self.attached.close()
84→ except Exception:
85→ pass
86→ if self.buffer:
87→ await ws.send_bytes(bytes(self.buffer)) # 行 87: 回放历史
88→ self.attached = ws
89→ self.last_active = time.time()
91→ async def detach(self, ws: WebSocket) -> None: # 行 91
92→ async with self.lock:
93→ if self.attached is ws:
94→ self.attached = None
95→ self.last_active = time.time()
97→ async def close(self) -> None: # 行 97
98→ if self.closed:
99→ return
100→ self.closed = True
101→ _SESSIONS.pop(self.id, None)
102→ cur = asyncio.current_task()
103→ if self.reader is not None and self.reader is not cur:
104→ self.reader.cancel()
105→ try:
106→ self.proc.close()
107→ except Exception:
108→ pass
109→ try:
110→ self.conn.close()
111→ except Exception:
112→ pass
行 122-128: 空闲清理定时任务
───────────────────────────────────────────────────────────────────
122→ async def _sweep_loop() -> None:
123→ while True:
124→ await asyncio.sleep(60) # 行 124
125→ now = time.time()
126→ for s in list(_SESSIONS.values()):
127→ if s.attached is None and now - s.last_active > _IDLE_TIMEOUT: # 行 127
128→ await s.close() # 行 128
行 138-151: terminal_open() 开启终端
───────────────────────────────────────────────────────────────────
138→ @router.post("/terminal/open")
139→ async def terminal_open(server: str):
140→ conn_kwargs, err = build_conn_kwargs(server) # 行 140
141→ if err:
142→ raise HTTPException(400, err)
143→ try:
144→ conn = await asyncssh.connect(**conn_kwargs) # 行 144
145→ proc = await conn.create_process( # 行 145
146→ term_type="xterm-256color",
147→ term_size=(80, 24),
148→ encoding=None
149→ )
150→ except Exception as e:
151→ raise HTTPException(400, f"连接失败:{type(e).__name__}: {e}") from None
152→ sid = uuid.uuid4().hex # 行 152
153→ _SESSIONS[sid] = TermSession(sid, server, conn, proc) # 行 153
154→ _ensure_sweeper() # 行 154
155→ return {"session_id": sid, "server": server} # 行 155
行 167-203: terminal_ws() WebSocket 处理
───────────────────────────────────────────────────────────────────
167→ @router.websocket("/terminal")
168→ async def terminal_ws(ws: WebSocket, session: str) -> None:
169→ await ws.accept()
170→ if not _ws_authed(ws): # 行 170
171→ await ws.send_json({"type": "error", ...})
172→ await ws.close()
173→ return
174→ sess = _SESSIONS.get(session)
175→ if sess is None or sess.closed:
176→ await ws.send_json({"type": "error", ...})
177→ await ws.close()
178→ return
179→ await sess.attach(ws) # 行 179
180→ await ws.send_json({"type": "ready", "server": sess.server}) # 行 181
181→ try:
182→ while True:
183→ msg = await ws.receive_json() # 行 184
184→ kind = msg.get("type")
185→ if kind == "input": # 行 186
186→ sess.proc.stdin.write(msg.get("data", "").encode()) # 行 187
187→ sess.last_active = time.time()
188→ elif kind == "resize": # 行 189
189→ sess.proc.change_terminal_size( # 行 191
190→ int(msg.get("cols", 80)),
191→ int(msg.get("rows", 24))
192→ )
193→ except WebSocketDisconnect:
194→ pass
195→ finally:
196→ await sess.detach(ws) # 行 199
197→ await ws.close()
───────────────────────────────────────────────────────────────────七、LLM 模型构建
文件: app/llm/registry.py
───────────────────────────────────────────────────────────────────
行 13-17: 默认 API 地址
───────────────────────────────────────────────────────────────────
13→ DEFAULT_BASE_URLS: dict[str, str] = {
14→ ProviderType.qwen: "https://dashscope.aliyuncs.com/compatible-mode/v1", # 行 14
15→ ProviderType.minimax: "https://api.minimaxi.com/v1", # 行 15
16→ ProviderType.deepseek: "https://api.deepseek.com/v1", # 行 16
17→ }
行 20-47: build_chat_model() 模型构建
───────────────────────────────────────────────────────────────────
20→ def build_chat_model(cfg: ModelProvider) -> BaseChatModel:
21→ api_key = decrypt(cfg.api_key_enc) # 行 21: 解密 API Key
22→ extra = cfg.extra or {}
23→
24→ if cfg.provider_type == ProviderType.anthropic: # 行 24
25→ from langchain_anthropic import ChatAnthropic # 行 25
26→
27→ return ChatAnthropic(
28→ model=cfg.model_name,
29→ api_key=api_key,
30→ base_url=cfg.base_url or None,
31→ temperature=cfg.temperature,
32→ stream_usage=True,
33→ **extra,
34→ )
36→ # openai / qwen / minimax / deepseek
37→ from langchain_openai import ChatOpenAI # 行 37
38→ base_url = cfg.base_url or DEFAULT_BASE_URLS.get(cfg.provider_type) # 行 39
39→ return ChatOpenAI(
40→ model=cfg.model_name,
41→ api_key=api_key,
42→ base_url=base_url, # 行 43
43→ temperature=cfg.temperature,
44→ stream_usage=True,
45→ **extra,
46→ )
───────────────────────────────────────────────────────────────────八、凭证加密
文件: app/db/crypto.py
───────────────────────────────────────────────────────────────────
行 14-16: Fernet 实例获取
───────────────────────────────────────────────────────────────────
14→ def _fernet() -> Fernet:
15→ key = get_settings().secret_encryption_key.encode()
16→ return Fernet(key)
行 19-28: 加密/解密函数
───────────────────────────────────────────────────────────────────
19→ def encrypt(plaintext: str | None) -> str | None:
20→ if plaintext is None or plaintext == "":
21→ return plaintext
22→ return _fernet().encrypt(plaintext.encode()).decode() # 行 22
23→
24→ def decrypt(ciphertext: str | None) -> str | None:
25→ if ciphertext is None or ciphertext == "":
26→ return ciphertext
27→ return _fernet().decrypt(ciphertext.encode()).decode() # 行 28
───────────────────────────────────────────────────────────────────九、数据模型
文件: app/db/models.py
───────────────────────────────────────────────────────────────────
行 40-52: ModelProvider 模型供应商
───────────────────────────────────────────────────────────────────
40→ class ModelProvider(Base):
41→ __tablename__ = "model_providers"
42→ id: Mapped[int] = mapped_column(primary_key=True)
43→ name: Mapped[str] = mapped_column(String(100), unique=True)
44→ provider_type: Mapped[ProviderType] = mapped_column(String(32))
45→ model_name: Mapped[str] = mapped_column(String(120))
46→ api_key_enc: Mapped[str | None] = mapped_column(Text, nullable=True) # 加密
47→ base_url: Mapped[str | None] = mapped_column(String(300), nullable=True)
48→ temperature: Mapped[float] = mapped_column(default=0.0)
49→ is_default: Mapped[bool] = mapped_column(Boolean, default=False)
50→ extra: Mapped[dict] = mapped_column(JSON, default=dict)
51→ created_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), default=_now)
行 58-67: SSHKey SSH 密钥库
───────────────────────────────────────────────────────────────────
58→ class SSHKey(Base):
59→ __tablename__ = "ssh_keys"
60→ id: Mapped[int] = mapped_column(primary_key=True)
61→ name: Mapped[str] = mapped_column(String(100), unique=True)
62→ private_key_enc: Mapped[str] = mapped_column(Text) # 加密
63→ passphrase_enc: Mapped[str | None] = mapped_column(Text, nullable=True) # 加密
64→ description: Mapped[str | None] = mapped_column(Text, nullable=True)
65→ created_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), default=_now)
行 70-86: Server 服务器
───────────────────────────────────────────────────────────────────
70→ class Server(Base):
71→ __tablename__ = "servers"
72→ id: Mapped[int] = mapped_column(primary_key=True)
73→ name: Mapped[str] = mapped_column(String(100), unique=True)
74→ host: Mapped[str] = mapped_column(String(255))
75→ port: Mapped[int] = mapped_column(Integer, default=22)
76→ username: Mapped[str] = mapped_column(String(100))
77→ auth_type: Mapped[str] = mapped_column(String(20), default="password")
78→ password_enc: Mapped[str | None] = mapped_column(Text, nullable=True) # 加密
79→ private_key_enc: Mapped[str | None] = mapped_column(Text, nullable=True) # 加密
80→ passphrase_enc: Mapped[str | None] = mapped_column(Text, nullable=True) # 加密
81→ ssh_key_id: Mapped[int | None] = mapped_column(ForeignKey("ssh_keys.id"), nullable=True)
82→ tags: Mapped[list] = mapped_column(JSON, default=list)
83→ description: Mapped[str | None] = mapped_column(Text, nullable=True)
84→ created_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), default=_now)
行 150-164: AuditLog 审计日志
───────────────────────────────────────────────────────────────────
150→ class AuditLog(Base):
151→ __tablename__ = "audit_logs"
152→ id: Mapped[int] = mapped_column(primary_key=True)
153→ conversation_id: Mapped[int | None] = mapped_column(ForeignKey("conversations.id"), nullable=True)
154→ tool_name: Mapped[str] = mapped_column(String(120))
155→ target: Mapped[str | None] = mapped_column(String(255), nullable=True)
156→ command: Mapped[str | None] = mapped_column(Text, nullable=True)
157→ level: Mapped[CommandLevel] = mapped_column(String(20), default=CommandLevel.readonly)
158→ approved: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
159→ approved_by: Mapped[str | None] = mapped_column(String(100), nullable=True)
160→ success: Mapped[bool] = mapped_column(Boolean, default=True)
161→ output: Mapped[str | None] = mapped_column(Text, nullable=True)
162→ created_at: Mapped[dt.datetime] = mapped_column(DateTime(timezone=True), default=_now)
163→ conversation: Mapped["Conversation | None"] = relationship(back_populates="audits")
164→ }
───────────────────────────────────────────────────────────────────十、完整调用流程汇总表
| 步骤 | 文件 | 行数 | 函数/类 | 说明 |
|---|---|---|---|---|
| 1 | main.py | 86-93 | run() | 应用启动入口 |
| 2 | main.py | 17-46 | lifespan() | 生命周期管理 |
| 3 | base.py | 28-32 | init_db() | 初始化数据库 |
| 4 | auth.py | 26-37 | login() | 用户登录 |
| 5 | crypto.py | 32-43 | sign_token/verify_token() | 会话令牌 |
| 6 | chat.py | 52-62 | chat_stream() | 发起对话 |
| 7 | runtime.py | 177-189 | astream_turn() | Agent 主入口 |
| 8 | runtime.py | 51-69 | _assemble() | 组装 Agent |
| 9 | runtime.py | 41-48 | _get_default_model() | 获取默认模型 |
| 10 | registry.py | 20-47 | build_chat_model() | 构建 LLM |
| 11 | ssh.py | 113-136 | make_scoped_ssh_tools() | 创建 SSH 工具 |
| 12 | runtime.py | 112-175 | _run_stream() | 流式事件处理 |
| 13 | graph.py | 38-229 | build_agent() | 构建 LangGraph |
| 14 | graph.py | 45-112 | agent_node() | LLM 调用 |
| 15 | graph.py | 115-162 | guardrail_node() | 命令分级+审批 |
| 16 | guardrails.py | 70-83 | classify_command_llm() | LLM 风险判断 |
| 17 | graph.py | 165-186 | execute_tools_node() | 执行工具 |
| 18 | ssh.py | 58-80 | _run_on_server() | SSH 执行命令 |
| 19 | ssh.py | 26-55 | build_conn_kwargs() | 构建连接参数 |
| 20 | crypto.py | 19-28 | encrypt/decrypt() | 凭证加解密 |
| 21 | graph.py | 252-263 | _audit() | 写入审计日志 |
| 22 | runtime.py | 192-203 | astream_resume() | 审批续跑 |
| 23 | terminal.py | 37-113 | TermSession | PTY 会话封装 |
| 24 | terminal.py | 138-155 | terminal_open() | 开启终端 |
| 25 | terminal.py | 167-203 | terminal_ws() | WebSocket 处理 |
以上就是详细的调用流程，每个函数/方法都标注了具体的代码文件名称和行号。如有疑问欢迎继续提问！