糖豆家

用户想了解项目启动后的类和方法调用过程。我需要详细梳理从项目启动到完整运行的整体流程，包括：

1. 应用启动入口 (main.py)
2. 数据库初始化
3. API 路由注册
4. Web 前端访问
5. 用户登录认证
6. Agent 对话流程
7. SSH 执行流程
8. 审批流程

让我详细分析这个调用链

一、应用启动流程

1.1 入口点 (`app/main.py`)

python -m app.main
       │
       ▼
run() 函数
  │
  ├─→ uvicorn.run("app.main:app", host, port, reload=True)
  │       │
  │       ▼
  │   FastAPI 应用实例化
  │       │
  │       ▼
  │   lifespan 上下文管理器执行
  │       │
  │       ├─→ init_db()                    # 初始化数据库
  │       │
  │       └─→ 模型配置验证
  │             ├─→ _get_default_model()   # 从 DB 读取默认模型
  │             └─→ build_chat_model()     # 创建模型实例
  │
  └─→ 注册路由
        ├─→ auth.router
        ├─→ admin.router
        ├─→ chat.router
        └─→ terminal.router

1.2 数据库初始化 (`app/db/base.py`)

python

init_db()
  │
  ├─→ import app.db.models        # 注册所有 ORM 模型
  │       │
  │       ├─→ Base.metadata.create_all()  # 创建表
  │       │
  │       └─→ _migrate_add_columns()      # 轻量迁移（补加新列）
  │
  └─→ 创建 SessionLocal 工厂

二、用户访问流程

2.1 Web 页面访问

浏览器访问 http://localhost:8000/
       │
       ▼
GET /
  │
  ├─→ 中间件检查 auth_enabled
  │       │
  │       └─→ 如果启用认证 → 验证 Cookie → 未登录返回 401
  │
  └─→ 返回 FileResponse("app/web/index.html")

2.2 前端文件 (`app/web/index.html`)

单文件前端（HTML + CSS + JS）
  │
  ├─→ 登录页面（未认证时）
  │       │
  │       └─→ POST /auth/login → 返回签名 Cookie
  │
  ├─→ 标签页导航
  │       ├─→ 模型配置
  │       ├─→ 服务器管理
  │       ├─→ SSH 密钥
  │       ├─→ 对话界面
  │       ├─→ 终端
  │       └─→ 审计日志
  │
  └─→ WebSocket 终端连接
          │
          └─→ ws://host/chat/terminal?session=xxx

三、用户登录认证流程

POST /auth/login {username, password}
       │
       ▼
login() [auth.py]
  │
  ├─→ secrets.compare_digest()    # 安全比较（防时序攻击）
  │
  ├─→ sign_token({"u": username, "t": timestamp})
  │       │
  │       └─→ Fernet 加密 + 签名
  │
  └─→ response.set_cookie("oa_session", token, ...)

后续请求认证中间件

请求进来
  │
  ▼
_auth_gate 中间件 [main.py]
  │
  ├─→ 检查路径是否放行（/auth/*, /, /health, /static/*）
  │
  └─→ verify_token(cookie["oa_session"], max_age)
          │
          └─→ Fernet 解密 + TTL 验证

四、Agent 对话核心流程

4.1 发起对话

POST /chat/stream {message, thread_id, servers, auto_approve_all}
       │
       ▼
chat_stream() [chat.py]
  │
  ├─→ 生成 thread_id（如果是新会话）
  │
  └─→ return _sse(astream_turn(...))
          │
          ▼
astream_turn() [runtime.py]
  │
  ├─→ _ensure_conversation()      # 创建/获取会话记录
  │
  ├─→ _save_message()             # 保存用户消息
  │
  ├─→ _assemble()                 # 组装 Agent
  │       │
  │       ├─→ _get_default_model()           # 获取模型配置
  │       ├─→ build_chat_model()             # 构建模型实例
  │       ├─→ make_scoped_ssh_tools()        # 创建 SSH 工具
  │       ├─→ load_cloud_tools()             # 加载 MCP 云工具
  │       │
  │       └─→ build_agent(model, tools, checkpointer)
  │               │
  │               └─→ 返回编译好的 LangGraph
  │
  └─→ agent.astream(graph_input, config)
          │
          ▼
┌─────────────────────────────────────────────────────────────────┐
│                    LangGraph 执行流程                           │
│                                                                 │
│  START ──▶ agent ──▶ [有 tool_calls?] ──▶ guardrail ──▶ execute │
│                      │                         │                 │
│                      │                         │ interrupt()    │
│                      │                         │ 等审批          │
│                      │                         ▼                 │
│                      │                      [批准/拒绝]         │
│                      │                         │                 │
│                      ◀─────────────────────────┘                 │
│                      │                                          │
│                      ▼                                          │
│                     END                                          │
└─────────────────────────────────────────────────────────────────┘

4.2 Agent 节点执行 (`graph.py`)

python

agent_node(state)
  │
  ├─→ 构建 SystemMessage（系统提示词）
  │
  ├─→ model_with_tools.ainvoke(messages)  # 调用 LLM
  │       │
  │       ├─→ 超时保护（60秒）
  │       │
  │       └─→ 异常分类处理
  │             ├─→ 402 余额不足
  │             ├─→ 401 认证失败
  │             └─→ 404 端点不存在
  │
  └─→ 返回 {"messages": [response], "last_io": {...}}

4.3 Guardrail 节点（命令分级）

python

guardrail_node(state)
  │
  ├─→ 检查 auto_approve_all
  │       │
  │       └─→ 全部放行
  │
  ├─→ 遍历 tool_calls
  │       │
  │       ├─→ needs_approval(tool_name)    # 是否需要审批
  │       │       │
  │       │       └─→ ssh_run 或云工具 __ 才需要
  │       │
  │       ├─→ is_auto_approved(command)    # 白名单检查
  │       │
  │       └─→ classify_command_llm()       # LLM 判断风险级别
  │               │
  │               └─→ readonly / mutating / dangerous
  │
  ├─→ 分类结果
  │       │
  │       ├─→ readonly → 自动放行
  │       ├─→ mutating → 需用户审批
  │       └─→ dangerous → 强制审批
  │
  └─→ interrupt({"type": "approval_required", "operations": [...]})
          │
          │ ⚠️ 流程在这里暂停！
          │ 前端收到 SSE 事件，等待用户点击"批准"
          ▼

4.4 工具执行节点

python

execute_tools_node(state)
  │
  ├─→ 获取 approved_ids（审批通过的 ID）
  │
  ├─→ 遍历 tool_calls
  │       │
  │       ├─→ 未批准 → 返回"[已跳过]"
  │       │
  │       └─→ 已批准 → tool.ainvoke(args)
  │               │
  │               ├─→ ssh_run → _run_on_server()
  │               └─→ 云工具 → MCP 调用
  │
  ├─→ _audit() 写入审计日志
  │
  └─→ 返回工具执行结果

五、SSH 命令执行流程

ssh_run_tool.ainvoke(args)
       │
       ▼
_run_on_server(server_name, command, intent, timeout)
  │
  ├─→ build_conn_kwargs(server_name)
  │       │
  │       ├─→ 从 DB 查询 Server 配置
  │       │
  │       ├─→ 解密凭证（password_enc / private_key_enc）
  │       │
  │       ├─→ 构建 asyncssh.connect 参数
  │       │       ├─→ host, port, username
  │       │       ├─→ password 或 client_keys
  │       │       └─→ known_hosts=None ⚠️（安全风险）
  │       │
  │       └─→ 返回 (conn_kwargs, error)
  │
  └─→ asyncssh.connect(**conn_kwargs)
          │
          ├─→ conn.run(command, check=False)
          │
          ├─→ 返回 stdout, stderr, exit_status
          │
          └─→ 格式化输出
                  │
                  └─→ "[exit=0]\nstdout:\n...\nstderr:\n..."

六、审批续跑流程

用户点击"批准" / "拒绝"
       │
       ▼
POST /chat/approve {thread_id, action, ids, remember, auto_approve_all}
       │
       ▼
approve() [chat.py]
  │
  └─→ return _sse(astream_resume(...))
          │
          ▼
astream_resume(thread_id, action, ids, remember, ...)
  │
  ├─→ _assemble()                 # 重新组装 Agent（保持状态）
  │
  ├─→ Command(resume={action, ids, remember}, 
  │           update={auto_approve_all})
  │       │
  │       └─→ LangGraph resume 恢复 interrupt 挂起的流程
  │
  └─→ _run_stream()               # 继续流式输出
          │
          ├─→ 收到 "__interrupt__" 的 payload
          │
          ├─→ 如果还有更多命令需要审批 → 再次 interrupt
          │
          └─→ 如果全部完成 → 返回 "done"

七、交互式终端流程

POST /chat/terminal/open?server=xxx
       │
       ▼
terminal_open() [terminal.py]
  │
  ├─→ build_conn_kwargs()         # 构建 SSH 连接参数
  │
  ├─→ asyncssh.connect()          # 建立连接
  │
  ├─→ conn.create_process()       # 创建 PTY shell
  │       │
  │       ├─→ term_type="xterm-256color"
  │       └─→ term_size=(80, 24)
  │
  ├─→ 创建 TermSession 对象
  │       │
  │       ├─→ 启动 _read_loop() 后台任务
  │       │
  │       └─→ 存入全局 _SESSIONS 字典
  │
  └─→ 返回 {session_id, server}
          │
          ▼
WebSocket 连接 ws://host/chat/terminal?session=xxx
       │
       ▼
terminal_ws() [terminal.py]
  │
  ├─→ 认证检查
  │
  ├─→ sess.attach(ws)             # 关联 WebSocket
  │
  ├─→ 发送历史缓冲数据（回放）
  │
  └─→ 循环处理消息
          │
          ├─→ {"type": "input", "data": "..."}
          │       │
          │       └─→ sess.proc.stdin.write()
          │
          └─→ {"type": "resize", "cols": 120, "rows": 30}
                  │
                  └─→ sess.proc.change_terminal_size()

终端会话生命周期

┌─────────────────────────────────────────────────────────────────┐
│                        TermSession                              │
├─────────────────────────────────────────────────────────────────┤
│  属性:                                                           │
│    - id, server, conn, proc                                     │
│    - buffer (256KB 回放缓冲)                                     │
│    - attached (当前 WebSocket)                                   │
│    - last_active (活动时间)                                      │
│                                                                 │
│  后台任务:                                                       │
│    - _read_loop(): 持续读取 PTY 输出                             │
│      └─→ 写入 buffer + 转发给 WebSocket                          │
│                                                                 │
│  定时清理:                                                       │
│    - _sweep_loop(): 每 60 秒检查                                 │
│      └─→ 30 分钟空闲 → close() 回收                             │
└─────────────────────────────────────────────────────────────────┘

八、完整调用时序图

┌──────┐     ┌─────────┐     ┌─────────┐     ┌────────┐     ┌──────────┐
│用户浏览器│     │FastAPI  │     │ Agent   │     │LangGraph│     │ SSH/云  │
└──────┘     └─────────┘     └─────────┘     └────────┘     └──────────┘
   │            │              │               │               │
   │ HTTP POST  │              │               │               │
   │ /chat/stream              │               │               │
   │───────────▶│              │               │               │
   │            │ astream_turn │               │               │
   │            │─────────────▶│               │               │
   │            │              │               │               │
   │            │              │ agent_node    │               │
   │            │              │──────────────▶│               │
   │            │              │               │ LLM 调用      │
   │            │              │               │───────────▶   │
   │            │              │               │               │
   │            │              │◀──────────────│ tool_calls    │
   │            │              │               │               │
   │            │              │ guardrail     │               │
   │            │              │──────────────▶│               │
   │            │              │               │ interrupt()   │
   │            │◀─────────────│ SSE 事件      │               │
   │ SSE 事件   │              │               │               │
   │◀───────────│              │               │               │
   │            │              │               │               │
   │ 显示审批框  │              │               │               │
   │            │              │               │               │
   │ 用户批准    │              │               │               │
   │───────────▶│              │               │               │
   │            │ astream_resume│              │               │
   │            │─────────────▶│               │               │
   │            │              │               │ resume        │
   │            │              │               │◀──────────────│
   │            │              │               │               │
   │            │              │ execute_tools │               │
   │            │              │──────────────▶│               │
   │            │              │               │ ssh_run       │
   │            │              │               │───────────▶   │
   │            │              │               │               │
   │            │              │               │◀──────────────│
   │            │              │◀──────────────│ 结果          │
   │            │◀─────────────│ SSE 事件      │               │
   │ SSE token  │              │               │               │
   │◀───────────│              │               │               │
   │            │              │               │               │
   │ 重复直到 END│              │               │               │

九、关键类和方法一览表

模块	类/函数	调用时机	作用
main.py	`run()`	启动时	启动 uvicorn
	`lifespan()`	启动/关闭	初始化 DB，验证模型
	`_auth_gate()`	每个请求	认证中间件
runtime.py	`astream_turn()`	用户发消息	发起 Agent 对话
	`astream_resume()`	审批通过	恢复中断的 Agent
	`_assemble()`	每次请求	组装 Agent 图
	`_run_stream()`	Agent 运行中	流式事件处理
graph.py	`build_agent()`	组装时	构建 LangGraph
	`agent_node()`	图执行	调用 LLM
	`guardrail_node()`	图执行	命令分级+审批
	`execute_tools_node()`	图执行	执行工具
guardrails.py	`classify_command()`	审批判断	正则判断风险级别
	`classify_command_llm()`	审批判断	LLM 判断风险级别
	`needs_approval()`	审批判断	是否需要审批
ssh.py	`_run_on_server()`	工具执行	SSH 实际执行命令
	`build_conn_kwargs()`	连接时	构建 SSH 参数
terminal.py	`TermSession`	终端创建	封装 PTY 会话
	`_sweep_loop()`	后台定时	清理空闲会话
crypto.py	`encrypt/decrypt()`	凭证读写	Fernet 加解密
	`sign_token/verify_token()`	登录/请求	会话令牌管理
models.py	各 ORM 模型	DB 操作	数据表映射

以上就是 Ops Agent 从启动到完整运行的全流程调用链。如有疑问可以继续提问！

一、应用启动流程 ​

1.1 入口点 (app/main.py) ​

1.2 数据库初始化 (app/db/base.py) ​

二、用户访问流程 ​

2.1 Web 页面访问 ​

2.2 前端文件 (app/web/index.html) ​

三、用户登录认证流程 ​

后续请求认证中间件 ​

四、Agent 对话核心流程 ​

4.1 发起对话 ​

4.2 Agent 节点执行 (graph.py) ​

4.3 Guardrail 节点（命令分级） ​

4.4 工具执行节点 ​

五、SSH 命令执行流程 ​

六、审批续跑流程 ​

七、交互式终端流程 ​

终端会话生命周期 ​

八、完整调用时序图 ​

九、关键类和方法一览表 ​

一、应用启动流程

1.1 入口点 (`app/main.py`)

1.2 数据库初始化 (`app/db/base.py`)

二、用户访问流程

2.1 Web 页面访问

2.2 前端文件 (`app/web/index.html`)

三、用户登录认证流程

后续请求认证中间件

四、Agent 对话核心流程

4.1 发起对话

4.2 Agent 节点执行 (`graph.py`)

4.3 Guardrail 节点（命令分级）

4.4 工具执行节点

五、SSH 命令执行流程

六、审批续跑流程

七、交互式终端流程

终端会话生命周期

八、完整调用时序图

九、关键类和方法一览表