1195 lines
39 KiB
Markdown
1195 lines
39 KiB
Markdown
# DeepAudit Agent 审计架构文档
|
||
|
||
## 目录
|
||
|
||
1. [系统概述](#1-系统概述)
|
||
2. [核心设计理念](#2-核心设计理念)
|
||
3. [后端架构](#3-后端架构)
|
||
- [Agent层级结构](#31-agent层级结构)
|
||
- [核心基础设施](#32-核心基础设施)
|
||
- [工具生态系统](#33-工具生态系统)
|
||
- [知识库与提示词](#34-知识库与提示词)
|
||
4. [前端架构](#4-前端架构)
|
||
5. [API接口](#5-api接口)
|
||
6. [审计任务执行流程](#6-审计任务执行流程)
|
||
7. [事件流与遥测](#7-事件流与遥测)
|
||
8. [配置与部署](#8-配置与部署)
|
||
9. [关键设计模式](#9-关键设计模式)
|
||
10. [安全与健壮性](#10-安全与健壮性)
|
||
11. [扩展指南](#11-扩展指南)
|
||
|
||
---
|
||
|
||
## 1. 系统概述
|
||
|
||
DeepAudit Agent 审计系统是一个基于大语言模型(LLM)驱动的自主化安全代码分析系统。该系统采用**动态多Agent层级架构**，实现了从代码侦察、漏洞分析到验证确认的完整安全审计流程。
|
||
|
||
### 核心特性
|
||
|
||
| 特性 | 描述 |
|
||
|------|------|
|
||
| **LLM中心化决策** | LLM作为系统的"大脑"，在各个层级自主做出决策 |
|
||
| **动态Agent树** | 根据任务需求动态构建层级化的多Agent系统 |
|
||
| **ReAct模式** | 实现思考-行动-观察(Thought-Action-Observation)循环 |
|
||
| **事件驱动架构** | 通过SSE实时推送事件到前端 |
|
||
| **丰富的工具生态** | 20+专业工具覆盖代码分析、模式匹配、漏洞验证 |
|
||
|
||
### 技术栈
|
||
|
||
- **后端**: Python 3.11+, FastAPI, LangChain/LiteLLM
|
||
- **前端**: React 18, TypeScript, Ant Design
|
||
- **数据库**: PostgreSQL (Supabase)
|
||
- **通信**: SSE (Server-Sent Events), REST API
|
||
|
||
---
|
||
|
||
## 2. 核心设计理念
|
||
|
||
### 2.1 LLM自主决策
|
||
|
||
与传统的规则驱动自动化不同，本系统中**LLM做出所有战略决策**：
|
||
|
||
```
|
||
传统方式: 规则 → 固定流程 → 结果
|
||
DeepAudit: LLM分析 → 自主决策 → 动态调整 → 结果
|
||
```
|
||
|
||
LLM负责决定：
|
||
- 何时派发哪个子Agent
|
||
- 使用哪些工具进行分析
|
||
- 何时认为分析已足够充分
|
||
- 如何解读和关联发现的问题
|
||
|
||
### 2.2 ReAct模式实现
|
||
|
||
每个Agent的执行遵循ReAct（Reasoning + Acting）模式：
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────┐
|
||
│ LLM 输出 │
|
||
├─────────────────────────────────────────────────┤
|
||
│ Thought: 我应该先了解项目结构... │
|
||
│ Action: list_files │
|
||
│ Action Input: {"directory": ".", "pattern": "*"}│
|
||
└─────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────┐
|
||
│ 系统执行工具 │
|
||
├─────────────────────────────────────────────────┤
|
||
│ Observation: [找到 app/, config/, tests/...] │
|
||
└─────────────────────────────────────────────────┘
|
||
↓
|
||
反馈给LLM，继续下一轮循环
|
||
```
|
||
|
||
### 2.3 层级化Agent协作
|
||
|
||
```
|
||
┌───────────────────┐
|
||
│ OrchestratorAgent │
|
||
│ (策略编排者) │
|
||
└─────────┬─────────┘
|
||
│
|
||
┌─────────────────┼─────────────────┐
|
||
↓ ↓ ↓
|
||
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
|
||
│ ReconAgent │ │ AnalysisAgent │ │VerificationAgent│
|
||
│ (情报收集) │ │ (漏洞猎手) │ │ (验证确认) │
|
||
└───────────────┘ └───────────────┘ └───────────────┘
|
||
```
|
||
|
||
---
|
||
|
||
## 3. 后端架构
|
||
|
||
### 3.1 Agent层级结构
|
||
|
||
所有Agent代码位于 `backend/app/services/agent/agents/` 目录。
|
||
|
||
#### 3.1.1 BaseAgent 基类
|
||
|
||
**文件**: `base.py`
|
||
|
||
BaseAgent是所有Agent的抽象基类，提供核心功能：
|
||
|
||
```python
|
||
class BaseAgent:
|
||
"""Agent基类 - 提供通用功能"""
|
||
|
||
# 核心配置
|
||
config: AgentConfig # 名称、类型、模式、最大token数、最大迭代次数
|
||
state: AgentState # 状态管理(status, messages, findings)
|
||
|
||
# 关键方法
|
||
async def run(self, input_data: dict) -> AgentResult # 抽象方法
|
||
async def stream_llm_call() # 统一的LLM调用(支持流式+自动压缩)
|
||
async def execute_tool() # 工具执行(带错误处理)
|
||
|
||
# 事件发射
|
||
def emit_thinking_token() # 流式输出LLM思考过程
|
||
def emit_thinking_start/end() # 思考生命周期
|
||
def emit_tool_call/result() # 工具执行生命周期
|
||
def emit_finding() # 漏洞发现事件
|
||
```
|
||
|
||
**关键特性**：
|
||
- 消息历史压缩（超过100k token时自动截断）
|
||
- 取消支持（`is_cancelled`属性）
|
||
- 知识模块动态加载
|
||
- 动态父子关系管理
|
||
|
||
#### 3.1.2 OrchestratorAgent 编排器
|
||
|
||
**文件**: `orchestrator.py`
|
||
**角色**: 战略编排与动态任务调度
|
||
**最大迭代次数**: 20
|
||
|
||
```python
|
||
class OrchestratorAgent(BaseAgent):
|
||
"""编排器Agent - 协调整体审计流程"""
|
||
|
||
async def run(self, input_data: dict) -> AgentResult:
|
||
"""主ReAct循环 - LLM决定下一步行动"""
|
||
# dispatch_agent: 派发子Agent
|
||
# summarize: 汇总当前进度
|
||
# finish: 完成审计
|
||
|
||
async def _dispatch_agent(self, agent_name: str, task: str):
|
||
"""派发ReconAgent、AnalysisAgent或VerificationAgent"""
|
||
|
||
def _parse_llm_response(self, response: str):
|
||
"""从LLM输出中提取Thought/Action/Action Input"""
|
||
```
|
||
|
||
**决策点**：
|
||
- 决定派发哪个子Agent
|
||
- 判断何时完成审计
|
||
- 汇总所有子Agent的发现
|
||
|
||
#### 3.1.3 ReconAgent 侦察员
|
||
|
||
**文件**: `recon.py`
|
||
**角色**: 项目分析与情报收集
|
||
**最大迭代次数**: 15
|
||
|
||
**主要任务**：
|
||
- 发现项目结构
|
||
- 识别技术栈和框架
|
||
- 定位入口点
|
||
- 标记高风险区域
|
||
|
||
**可用工具**: `list_files`, `read_file`, `search_code`
|
||
|
||
**输出格式**：
|
||
```json
|
||
{
|
||
"project_structure": {...},
|
||
"tech_stack": {"languages": ["Python"], "frameworks": ["Django"]},
|
||
"entry_points": ["app/views.py", "api/routes.py"],
|
||
"high_risk_areas": ["auth/", "payment/"],
|
||
"initial_findings": [...]
|
||
}
|
||
```
|
||
|
||
#### 3.1.4 AnalysisAgent 分析员
|
||
|
||
**文件**: `analysis.py`
|
||
**角色**: 深度代码漏洞分析
|
||
**最大迭代次数**: 30
|
||
|
||
**主要工具**：
|
||
|
||
| 工具 | 描述 | 优先级 |
|
||
|------|------|--------|
|
||
| `smart_scan` | 智能批量安全扫描 | **推荐首选** |
|
||
| `quick_audit` | 快速文件审计 | 二级 |
|
||
| `pattern_match` | 危险模式检测 | 三级 |
|
||
| `dataflow_analysis` | 数据流追踪 | 深度分析 |
|
||
| `semgrep_scan` | Semgrep静态分析 | 外部工具 |
|
||
| `bandit_scan` | Python安全检测 | 外部工具 |
|
||
| `gitleaks_scan` | 密钥泄露检测 | 外部工具 |
|
||
|
||
**关注的漏洞类型**：
|
||
- SQL注入
|
||
- XSS跨站脚本
|
||
- 命令注入
|
||
- 路径遍历
|
||
- SSRF服务端请求伪造
|
||
|
||
#### 3.1.5 VerificationAgent 验证员
|
||
|
||
**文件**: `verification.py`
|
||
**角色**: 确认发现并生成PoC
|
||
**最大迭代次数**: 15
|
||
|
||
**主要职责**：
|
||
- 减少误报
|
||
- 验证可利用性
|
||
- 生成概念验证(PoC)
|
||
|
||
**输出格式**：
|
||
```json
|
||
{
|
||
"is_verified": true,
|
||
"poc": {
|
||
"description": "...",
|
||
"steps": [...],
|
||
"payload": "..."
|
||
},
|
||
"exploitability": "high",
|
||
"confidence": 0.95
|
||
}
|
||
```
|
||
|
||
#### 3.1.6 TaskHandoff 任务交接协议
|
||
|
||
Agent之间通过结构化的`TaskHandoff`进行协作：
|
||
|
||
```python
|
||
@dataclass
|
||
class TaskHandoff:
|
||
"""Agent间的结构化通信协议"""
|
||
work_completed: str # 已完成的工作
|
||
key_findings: List[dict] # 关键发现
|
||
insights: List[str] # 洞察
|
||
suggested_actions: List[str] # 建议行动
|
||
attention_points: List[str] # 需要关注的点
|
||
priority_areas: List[str] # 优先区域
|
||
context_data: dict # 上下文数据
|
||
|
||
def to_prompt_context(self) -> str:
|
||
"""转换为LLM可读格式"""
|
||
```
|
||
|
||
---
|
||
|
||
### 3.2 核心基础设施
|
||
|
||
核心基础设施位于 `backend/app/services/agent/core/` 目录。
|
||
|
||
#### 3.2.1 状态管理 (state.py)
|
||
|
||
```python
|
||
class AgentState:
|
||
"""Agent状态跟踪"""
|
||
status: Literal["created", "running", "waiting", "completed", "failed"]
|
||
messages: List[Message] # 对话历史
|
||
findings: List[Finding] # 发现列表
|
||
iterations: int # 当前迭代次数
|
||
tool_calls: int # 工具调用次数
|
||
```
|
||
|
||
#### 3.2.2 执行上下文 (context.py)
|
||
|
||
```python
|
||
class ExecutionContext:
|
||
"""分布式追踪的执行上下文"""
|
||
task_id: str # 任务ID
|
||
agent_id: str # Agent ID
|
||
trace_path: str # 追踪路径 (如: Orchestrator > Analysis)
|
||
depth: int # 嵌套深度
|
||
iteration: int # 当前迭代
|
||
metadata: dict # 附加元数据
|
||
|
||
def child_context(self, agent_id: str) -> ExecutionContext:
|
||
"""创建子上下文"""
|
||
|
||
def with_iteration(self, iteration: int) -> ExecutionContext:
|
||
"""带迭代信息的上下文"""
|
||
```
|
||
|
||
#### 3.2.3 熔断器 (circuit_breaker.py)
|
||
|
||
```python
|
||
class CircuitBreaker:
|
||
"""熔断器 - 防止级联故障"""
|
||
|
||
# 状态转换: CLOSED -> OPEN -> HALF_OPEN -> CLOSED
|
||
|
||
failure_threshold: int = 5 # 失败阈值
|
||
recovery_timeout: float = 30.0 # 恢复超时(秒)
|
||
|
||
async def call(self, func: Callable) -> Any:
|
||
"""受保护的调用"""
|
||
if self.state == CircuitState.OPEN:
|
||
raise CircuitOpenError()
|
||
try:
|
||
result = await func()
|
||
self._record_success()
|
||
return result
|
||
except Exception as e:
|
||
self._record_failure()
|
||
raise
|
||
```
|
||
|
||
#### 3.2.4 重试机制 (retry.py)
|
||
|
||
```python
|
||
class RetryStrategy:
|
||
"""指数退避重试"""
|
||
base_delay: float = 1.0 # 基础延迟(秒)
|
||
max_delay: float = 60.0 # 最大延迟(秒)
|
||
max_retries: int = 3 # 最大重试次数
|
||
|
||
def get_delay(self, attempt: int) -> float:
|
||
"""计算延迟: min(base * (2 ** attempt) + jitter, max)"""
|
||
```
|
||
|
||
#### 3.2.5 速率限制器 (rate_limiter.py)
|
||
|
||
```python
|
||
class RateLimiter:
|
||
"""令牌桶算法速率限制"""
|
||
|
||
# 预设限制:
|
||
# - 外部工具: 0.2 calls/second (每5秒1次)
|
||
# - LLM调用: 60 calls/minute
|
||
# - 突发支持: 最多3个并发调用
|
||
|
||
async def acquire(self, tool_name: str):
|
||
"""获取执行许可"""
|
||
```
|
||
|
||
#### 3.2.6 输入验证 (validation.py)
|
||
|
||
```python
|
||
class InputValidator:
|
||
"""输入安全验证"""
|
||
|
||
def validate_file_path(self, path: str) -> bool:
|
||
"""文件路径安全检查"""
|
||
# - 检查路径遍历攻击
|
||
# - 验证文件扩展名
|
||
# - 检查符号链接
|
||
|
||
def validate_file_size(self, path: str, max_size: int = 10_000_000):
|
||
"""文件大小限制(默认10MB)"""
|
||
```
|
||
|
||
#### 3.2.7 错误处理 (errors.py)
|
||
|
||
```python
|
||
# 自定义异常层级
|
||
class AgentError(Exception): pass
|
||
class CircuitOpenError(AgentError): pass # 熔断器打开
|
||
class AgentExecutionError(AgentError): pass # Agent执行失败
|
||
class ToolExecutionError(AgentError): pass # 工具执行失败
|
||
class ValidationError(AgentError): pass # 输入验证失败
|
||
class TokenLimitError(AgentError): pass # Token超限
|
||
```
|
||
|
||
---
|
||
|
||
### 3.3 工具生态系统
|
||
|
||
工具代码位于 `backend/app/services/agent/tools/` 目录。
|
||
|
||
#### 3.3.1 文件操作工具
|
||
|
||
| 工具 | 文件 | 描述 |
|
||
|------|------|------|
|
||
| `FileReadTool` | - | 读取文件内容(支持行范围) |
|
||
| `ListFilesTool` | - | 列出目录内容(支持glob模式) |
|
||
| `FileSearchTool` | - | 按关键词搜索代码 |
|
||
|
||
#### 3.3.2 代码分析工具
|
||
|
||
**SmartScanTool** (`smart_scan_tool.py`)
|
||
```python
|
||
class SmartScanTool:
|
||
"""智能批量安全扫描 - 推荐首选工具"""
|
||
|
||
# 功能:
|
||
# - 自动检测漏洞模式
|
||
# - 聚焦高风险文件
|
||
# - 批量处理提高效率
|
||
|
||
async def execute(self, target: str) -> dict:
|
||
return {
|
||
"vulnerabilities": [...],
|
||
"risk_areas": [...],
|
||
"recommendations": [...]
|
||
}
|
||
```
|
||
|
||
**PatternMatchTool** (`pattern_tool.py`)
|
||
```python
|
||
class PatternMatchTool:
|
||
"""基于正则的模式检测"""
|
||
|
||
# 内置模式:
|
||
# - SQL注入模式
|
||
# - XSS模式
|
||
# - 命令注入模式
|
||
# - 可作为Semgrep的后备
|
||
```
|
||
|
||
**DataFlowAnalysisTool** (`code_analysis_tool.py`)
|
||
```python
|
||
class DataFlowAnalysisTool:
|
||
"""数据流分析工具"""
|
||
|
||
# 功能:
|
||
# - 源点到汇点追踪
|
||
# - 变量污点分析
|
||
# - LLM辅助的数据流理解
|
||
```
|
||
|
||
#### 3.3.3 外部安全工具
|
||
|
||
**文件**: `external_tools.py`
|
||
|
||
| 工具 | 描述 | 超时 | 后备方案 |
|
||
|------|------|------|----------|
|
||
| `SemgrepTool` | 静态分析规则引擎 | 120s | PatternMatchTool |
|
||
| `BanditTool` | Python安全linter | 60s | PatternMatchTool |
|
||
| `GitleaksTool` | 密钥/凭证检测 | 60s | - |
|
||
|
||
#### 3.3.4 完成工具
|
||
|
||
**FinishTool** (`finish_tool.py`)
|
||
```python
|
||
class FinishTool:
|
||
"""任务完成工具"""
|
||
|
||
async def execute(self, conclusion: str, findings: List[dict]):
|
||
"""标记任务完成并返回最终结果"""
|
||
```
|
||
|
||
---
|
||
|
||
### 3.4 知识库与提示词
|
||
|
||
#### 3.4.1 漏洞知识库
|
||
|
||
位于 `backend/app/services/agent/knowledge/vulnerabilities/`
|
||
|
||
**按漏洞类型组织**：
|
||
|
||
| 文件 | 漏洞类型 |
|
||
|------|----------|
|
||
| `sql_injection.py` | SQL注入 |
|
||
| `xss.py` | 跨站脚本 |
|
||
| `csrf.py` | 跨站请求伪造 |
|
||
| `auth.py` | 认证漏洞 |
|
||
| `ssrf.py` | 服务端请求伪造 |
|
||
| `path_traversal.py` | 路径遍历 |
|
||
| `deserialization.py` | 不安全反序列化 |
|
||
| `xxe.py` | XML外部实体注入 |
|
||
| `race_condition.py` | 竞态条件 |
|
||
| `crypto.py` | 弱加密 |
|
||
| `injection.py` | 代码/命令注入 |
|
||
| `business_logic.py` | 业务逻辑漏洞 |
|
||
| `open_redirect.py` | 开放重定向 |
|
||
|
||
**框架特定知识**：
|
||
|
||
| 文件 | 框架 |
|
||
|------|------|
|
||
| `FastAPI.py` | FastAPI安全模式 |
|
||
| `Django.py` | Django安全问题 |
|
||
| `Flask.py` | Flask漏洞 |
|
||
| `Express.js` | Node.js/Express模式 |
|
||
| `React.js` | 前端安全 |
|
||
| `Supabase.py` | BaaS安全 |
|
||
|
||
#### 3.4.2 系统提示词
|
||
|
||
**文件**: `backend/app/services/agent/prompts/system_prompts.py`
|
||
|
||
```python
|
||
# 核心安全原则
|
||
CORE_SECURITY_PRINCIPLES = """
|
||
- 深度分析优于广度覆盖
|
||
- 重视数据流追踪
|
||
- 上下文感知分析
|
||
- 假阳性需验证确认
|
||
"""
|
||
|
||
# 漏洞优先级
|
||
VULNERABILITY_PRIORITIES = {
|
||
"critical": ["sql_injection", "command_injection", "code_injection"],
|
||
"high": ["path_traversal", "ssrf", "auth_bypass"],
|
||
"medium": ["xss", "information_disclosure", "xxe"],
|
||
"low": ["csrf", "weak_crypto", "unsafe_transport"]
|
||
}
|
||
|
||
# 多Agent协作规则
|
||
MULTI_AGENT_RULES = """
|
||
- 避免重复工作
|
||
- 共享上下文信息
|
||
- 聚焦自身职责
|
||
- 及时交接发现
|
||
"""
|
||
```
|
||
|
||
---
|
||
|
||
## 4. 前端架构
|
||
|
||
前端代码位于 `frontend/src/pages/AgentAudit/` 目录。
|
||
|
||
### 4.1 目录结构
|
||
|
||
```
|
||
AgentAudit/
|
||
├── index.tsx # 主页面组件
|
||
├── types.ts # TypeScript类型定义
|
||
├── utils.ts # 工具函数
|
||
├── constants.tsx # 常量配置
|
||
├── hooks/
|
||
│ ├── useAgentAuditState.ts # 状态管理(useReducer)
|
||
│ ├── useResilientStream.ts # SSE流式连接
|
||
│ └── index.ts
|
||
└── components/
|
||
├── Header.tsx # 标题、状态、控制按钮
|
||
├── StatsPanel.tsx # 统计面板
|
||
├── AgentTreeNode.tsx # Agent树节点渲染
|
||
├── AgentDetailPanel.tsx # Agent详情侧边栏
|
||
├── LogEntry.tsx # 单条日志条目
|
||
├── ConnectionStatus.tsx # 连接状态指示器
|
||
├── StatusBadge.tsx # 状态徽章
|
||
├── SplashScreen.tsx # 初始屏幕
|
||
├── AgentErrorBoundary.tsx # 错误边界
|
||
├── ReportExportDialog.tsx # 导出对话框
|
||
└── index.ts
|
||
```
|
||
|
||
### 4.2 状态管理
|
||
|
||
**文件**: `hooks/useAgentAuditState.ts`
|
||
|
||
```typescript
|
||
interface AgentAuditState {
|
||
task: AgentTask | null; // 当前任务
|
||
findings: AgentFinding[]; // 发现列表
|
||
agentTree: AgentTreeResponse | null; // Agent树
|
||
logs: LogItem[]; // 日志列表
|
||
selectedAgentId: string | null; // 选中的Agent
|
||
connectionStatus: ConnectionStatus; // 连接状态
|
||
isAutoScroll: boolean; // 自动滚动
|
||
expandedLogIds: Set<string>; // 展开的日志ID
|
||
}
|
||
|
||
type ConnectionStatus = 'disconnected' | 'connecting' | 'connected' | 'error';
|
||
|
||
// Reducer Actions
|
||
type Action =
|
||
| { type: 'SET_TASK'; payload: AgentTask }
|
||
| { type: 'SET_FINDINGS'; payload: AgentFinding[] }
|
||
| { type: 'ADD_LOG'; payload: LogItem }
|
||
| { type: 'UPDATE_LOG'; payload: { id: string; updates: Partial<LogItem> } }
|
||
| { type: 'SET_CONNECTION_STATUS'; payload: ConnectionStatus }
|
||
| { type: 'SELECT_AGENT'; payload: string | null }
|
||
| { type: 'TOGGLE_LOG_EXPANDED'; payload: string }
|
||
| { type: 'RESET' };
|
||
```
|
||
|
||
### 4.3 事件流处理
|
||
|
||
**文件**: `hooks/useResilientStream.ts`
|
||
|
||
```typescript
|
||
function useResilientStream(taskId: string, options: StreamOptions) {
|
||
// 功能:
|
||
// - 弹性SSE连接(自动重连)
|
||
// - Thinking token逐字符流式更新
|
||
// - 按类型过滤事件
|
||
// - 组件卸载时自动清理
|
||
|
||
// 事件处理器
|
||
onThinkingToken: (token: string) => void; // 累积thinking内容
|
||
onToolStart: (tool: ToolEvent) => void; // 工具开始
|
||
onToolEnd: (tool: ToolEvent) => void; // 工具结束
|
||
onFinding: (finding: Finding) => void; // 新发现
|
||
onComplete: () => void; // 任务完成
|
||
onError: (error: Error) => void; // 错误处理
|
||
}
|
||
```
|
||
|
||
### 4.4 日志类型
|
||
|
||
```typescript
|
||
type LogType =
|
||
| 'thinking' // LLM思考过程(支持流式)
|
||
| 'tool' // 工具执行(含耗时和状态)
|
||
| 'phase' // 工作流阶段
|
||
| 'finding' // 漏洞发现
|
||
| 'info' // 信息消息
|
||
| 'error' // 错误消息
|
||
| 'dispatch' // 子Agent派发
|
||
| 'user'; // 用户操作
|
||
|
||
interface LogItem {
|
||
id: string;
|
||
type: LogType;
|
||
timestamp: Date;
|
||
content: string;
|
||
metadata?: Record<string, any>;
|
||
isExpanded?: boolean;
|
||
isStreaming?: boolean; // thinking类型专用
|
||
}
|
||
```
|
||
|
||
### 4.5 实时更新流程
|
||
|
||
```
|
||
SSE Stream → useResilientStream → dispatch() → reducer → UI更新
|
||
↓
|
||
thinking_token事件
|
||
↓
|
||
onThinkingToken回调
|
||
↓
|
||
dispatch({type: 'ADD_LOG', payload: {type: 'thinking', content: accumulated}})
|
||
↓
|
||
UI渲染LogEntry(流式效果)
|
||
```
|
||
|
||
---
|
||
|
||
## 5. API接口
|
||
|
||
API端点位于 `backend/app/api/v1/endpoints/agent_tasks.py`
|
||
|
||
### 5.1 任务管理
|
||
|
||
| 方法 | 端点 | 描述 |
|
||
|------|------|------|
|
||
| POST | `/agent-tasks/` | 创建并启动审计任务 |
|
||
| GET | `/agent-tasks/` | 列出任务(支持过滤) |
|
||
| GET | `/agent-tasks/{task_id}` | 获取任务详情 |
|
||
| POST | `/agent-tasks/{task_id}/cancel` | 取消运行中的任务 |
|
||
|
||
### 5.2 事件流
|
||
|
||
| 方法 | 端点 | 描述 |
|
||
|------|------|------|
|
||
| GET | `/agent-tasks/{task_id}/events` | 基础事件轮询(SSE) |
|
||
| GET | `/agent-tasks/{task_id}/stream` | 增强事件流(含thinking tokens) |
|
||
|
||
**查询参数**：
|
||
- `include_thinking`: 是否包含thinking token
|
||
- `include_tool_calls`: 是否包含工具调用
|
||
- `after_sequence`: 从指定序列号后开始
|
||
|
||
### 5.3 结果查询
|
||
|
||
| 方法 | 端点 | 描述 |
|
||
|------|------|------|
|
||
| GET | `/agent-tasks/{task_id}/findings` | 获取发现列表(支持分页) |
|
||
| GET | `/agent-tasks/{task_id}/summary` | 任务摘要与统计 |
|
||
| PATCH | `/agent-tasks/{task_id}/findings/{finding_id}/status` | 更新发现状态 |
|
||
| GET | `/agent-tasks/{task_id}/report` | 生成报告(markdown/json) |
|
||
|
||
### 5.4 高级功能
|
||
|
||
| 方法 | 端点 | 描述 |
|
||
|------|------|------|
|
||
| GET | `/agent-tasks/{task_id}/agent-tree` | Agent层级与执行统计 |
|
||
| GET | `/agent-tasks/{task_id}/checkpoints` | 执行检查点列表 |
|
||
| GET | `/agent-tasks/{task_id}/checkpoints/{checkpoint_id}` | 检查点详情 |
|
||
|
||
### 5.5 创建任务Schema
|
||
|
||
```python
|
||
class AgentTaskCreate(BaseModel):
|
||
project_id: str # 项目ID
|
||
name: Optional[str] # 任务名称
|
||
target_vulnerabilities: List[str] = [
|
||
"sql_injection", "xss", "command_injection",
|
||
"path_traversal", "ssrf"
|
||
]
|
||
verification_level: str = "sandbox" # 验证级别
|
||
exclude_patterns: List[str] = [ # 排除模式
|
||
"node_modules", "venv", "__pycache__", ".git"
|
||
]
|
||
target_files: Optional[List[str]] # 限定扫描文件
|
||
max_iterations: int = 50 # 最大迭代(1-200)
|
||
timeout_seconds: int = 1800 # 超时(60-7200秒)
|
||
```
|
||
|
||
---
|
||
|
||
## 6. 审计任务执行流程
|
||
|
||
### 6.1 完整流程图
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 1. 任务创建 │
|
||
│ 用户 → POST /agent-tasks/ → 创建AgentTask(PENDING) → 入队后台任务 │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 2. 初始化 │
|
||
│ 加载项目 → 收集项目信息 → 初始化工具 → 创建子Agent → 创建Orchestrator │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 3. Orchestrator循环 (最多20轮) │
|
||
│ ┌──────────────────────────────────────────────────────────────┐ │
|
||
│ │ 迭代 N: │ │
|
||
│ │ a) 调用LLM获取下一步决策 │ │
|
||
│ │ b) 解析: Thought / Action / Action Input │ │
|
||
│ │ c) 执行Action: │ │
|
||
│ │ - dispatch_agent → 派发子Agent │ │
|
||
│ │ - finish → 完成审计 │ │
|
||
│ │ d) 发射事件到前端 │ │
|
||
│ └──────────────────────────────────────────────────────────────┘ │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 4. 子Agent执行 (以Analysis为例) │
|
||
│ ┌──────────────────────────────────────────────────────────────┐ │
|
||
│ │ 循环 (最多30轮): │ │
|
||
│ │ a) LLM思考下一步 │ │
|
||
│ │ b) 选择并执行工具 (smart_scan, pattern_match, ...) │ │
|
||
│ │ c) 将观察结果加入历史 │ │
|
||
│ │ d) 检测是否输出"Final Answer" → 解析发现，返回 │ │
|
||
│ └──────────────────────────────────────────────────────────────┘ │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 5. 结果汇总 │
|
||
│ 合并所有发现 → 去重 → 更新验证结果 → 构建findings列表 │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 6. 持久化 │
|
||
│ 更新AgentTask状态 → 保存到AgentFinding表 → 计算安全评分 │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
↓
|
||
┌─────────────────────────────────────────────────────────────────────┐
|
||
│ 7. 前端展示 │
|
||
│ 实时SSE事件 → 日志累积 → Agent树生长 → 统计更新 → 最终结果展示 │
|
||
└─────────────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
### 6.2 详细执行示例
|
||
|
||
**场景**: 用户上传Python项目，执行Agent审计
|
||
|
||
```
|
||
1. 用户请求: POST /agent-tasks/
|
||
{
|
||
"project_id": "123",
|
||
"target_files": ["app/views.py"]
|
||
}
|
||
|
||
2. Orchestrator迭代1 - LLM输出:
|
||
Thought: 我应该先用recon了解项目结构
|
||
Action: dispatch_agent
|
||
Action Input: {"agent": "recon", "task": "分析项目结构"}
|
||
|
||
3. ReconAgent执行:
|
||
迭代1: list_files → 找到 app/views.py, models.py...
|
||
迭代2: read_file → 读取app/views.py内容
|
||
迭代N: Final Answer → 返回项目结构信息
|
||
|
||
4. Orchestrator迭代2 - LLM输出:
|
||
Thought: 这是一个Django项目，让我派发Analysis进行漏洞分析
|
||
Action: dispatch_agent
|
||
Action Input: {"agent": "analysis", "task": "深度漏洞分析"}
|
||
|
||
5. AnalysisAgent执行:
|
||
迭代1: smart_scan → 发现第42行可能存在SQL注入
|
||
迭代2: read_file → 确认问题代码
|
||
迭代M: Final Answer → 返回结构化的漏洞发现
|
||
|
||
6. Orchestrator迭代3 - LLM输出:
|
||
Thought: 发现高危SQL注入，需要验证
|
||
Action: dispatch_agent
|
||
Action Input: {"agent": "verification", "task": "验证SQL注入"}
|
||
|
||
7. VerificationAgent确认:
|
||
返回: {"is_verified": true, "poc": {...}}
|
||
|
||
8. Orchestrator完成:
|
||
Thought: 审计完成，共发现1个已验证漏洞
|
||
Action: finish
|
||
Action Input: {"conclusion": "...", "findings": [...]}
|
||
|
||
9. 后端持久化:
|
||
- AgentTask.status = COMPLETED
|
||
- 写入AgentFinding表
|
||
|
||
10. 前端实时展示:
|
||
- 日志流式更新
|
||
- Agent树动态生长
|
||
- 统计数据刷新
|
||
- 最终展示发现列表
|
||
```
|
||
|
||
---
|
||
|
||
## 7. 事件流与遥测
|
||
|
||
### 7.1 事件管理器
|
||
|
||
**文件**: `backend/app/services/agent/event_manager.py`
|
||
|
||
```python
|
||
class EventManager:
|
||
"""事件管理器 - 内存队列 + 流式迭代"""
|
||
|
||
max_events: int = 1000 # 最大事件数
|
||
|
||
async def emit(self, event_type: str, message: str, **metadata):
|
||
"""发射事件"""
|
||
|
||
async def stream_events(self):
|
||
"""异步迭代器: async for event in manager.stream_events()"""
|
||
```
|
||
|
||
### 7.2 事件类型
|
||
|
||
| 类型 | 描述 |
|
||
|------|------|
|
||
| `thinking_start` | LLM开始思考 |
|
||
| `thinking_token` | 思考token(逐字符) |
|
||
| `thinking_end` | LLM思考结束 |
|
||
| `tool_call` | 工具调用开始 |
|
||
| `tool_result` | 工具执行结果 |
|
||
| `finding` | 漏洞发现 |
|
||
| `phase_start` | 阶段开始 |
|
||
| `phase_complete` | 阶段完成 |
|
||
| `dispatch` | 子Agent派发 |
|
||
| `llm_thought` | LLM思考内容 |
|
||
| `llm_decision` | LLM决策 |
|
||
| `error` | 错误 |
|
||
| `warning` | 警告 |
|
||
| `info` | 信息 |
|
||
|
||
### 7.3 遥测追踪
|
||
|
||
**文件**: `backend/app/services/agent/telemetry/tracer.py`
|
||
|
||
```python
|
||
class Tracer:
|
||
"""分布式追踪"""
|
||
|
||
def create_span(self, name: str, parent_span: Span = None) -> Span:
|
||
"""创建追踪Span"""
|
||
|
||
# 追踪路径示例: Orchestrator > Analysis > (Verification)
|
||
|
||
# 记录的指标:
|
||
# - 执行时间
|
||
# - Token使用量
|
||
# - 工具调用次数
|
||
# - 关联ID
|
||
```
|
||
|
||
---
|
||
|
||
## 8. 配置与部署
|
||
|
||
### 8.1 Agent配置
|
||
|
||
**文件**: `backend/app/services/agent/config.py`
|
||
|
||
```python
|
||
class AgentConfig:
|
||
"""Agent配置 (环境变量前缀: AGENT_)"""
|
||
|
||
# LLM设置
|
||
llm_max_retries: int = 3
|
||
llm_retry_base_delay: float = 1.0
|
||
llm_timeout_seconds: int = 120
|
||
llm_max_tokens_per_call: int = 4096
|
||
llm_temperature: float = 0.1
|
||
llm_stream_enabled: bool = True
|
||
|
||
# Agent迭代限制
|
||
orchestrator_max_iterations: int = 20
|
||
recon_max_iterations: int = 15
|
||
analysis_max_iterations: int = 30
|
||
verification_max_iterations: int = 15
|
||
|
||
# 工具设置
|
||
tool_timeout_seconds: int = 60
|
||
tool_max_retries: int = 2
|
||
semgrep_enabled: bool = True
|
||
bandit_enabled: bool = True
|
||
gitleaks_enabled: bool = True
|
||
|
||
# 资源限制
|
||
max_file_size_bytes: int = 10_000_000 # 10MB
|
||
max_files_per_scan: int = 1000
|
||
max_total_findings: int = 500
|
||
max_context_messages: int = 50
|
||
|
||
# 熔断器
|
||
circuit_breaker_enabled: bool = True
|
||
circuit_failure_threshold: int = 5
|
||
circuit_recovery_timeout_seconds: float = 30.0
|
||
|
||
# 检查点
|
||
checkpoint_enabled: bool = True
|
||
checkpoint_interval_iterations: int = 5
|
||
max_checkpoints_per_task: int = 50
|
||
```
|
||
|
||
### 8.2 环境预设
|
||
|
||
```python
|
||
def apply_development_preset():
|
||
"""开发环境预设"""
|
||
# 更宽松的限制，更多日志
|
||
|
||
def apply_production_preset():
|
||
"""生产环境预设"""
|
||
# 更严格的限制，优化性能
|
||
|
||
def apply_testing_preset():
|
||
"""测试环境预设"""
|
||
# 快速迭代，最小资源
|
||
```
|
||
|
||
---
|
||
|
||
## 9. 关键设计模式
|
||
|
||
### 9.1 LLM中心化决策
|
||
|
||
```
|
||
┌─────────────────────────────────────────────┐
|
||
│ 传统自动化 │
|
||
│ 规则引擎 → 固定流程 → 预定义行为 │
|
||
└─────────────────────────────────────────────┘
|
||
|
||
┌─────────────────────────────────────────────┐
|
||
│ DeepAudit │
|
||
│ LLM分析 → 自主决策 → 动态适应 → 智能输出 │
|
||
└─────────────────────────────────────────────┘
|
||
```
|
||
|
||
### 9.2 ReAct模式
|
||
|
||
```
|
||
Loop:
|
||
1. Thought: LLM思考当前状态和下一步
|
||
2. Action: 选择要执行的动作/工具
|
||
3. Action Input: 提供参数
|
||
4. Observation: 获取执行结果
|
||
5. 重复直到达成目标或Final Answer
|
||
```
|
||
|
||
### 9.3 任务交接协议
|
||
|
||
```python
|
||
# Agent A完成工作后
|
||
handoff = TaskHandoff(
|
||
work_completed="扫描了10个文件，发现3个可疑点",
|
||
key_findings=[finding1, finding2, finding3],
|
||
suggested_actions=["验证SQL注入", "检查认证逻辑"],
|
||
priority_areas=["auth/", "api/"]
|
||
)
|
||
|
||
# 转换为下一个Agent的输入
|
||
context = handoff.to_prompt_context()
|
||
# Agent B使用此上下文继续工作
|
||
```
|
||
|
||
### 9.4 优雅降级
|
||
|
||
```
|
||
┌───────────────────────────────────────────────────┐
|
||
│ 优雅降级策略 │
|
||
├───────────────────────────────────────────────────┤
|
||
│ • 工具失败不中断执行 (continue_on_tool_failure) │
|
||
│ • 后备工具链 (Semgrep → PatternMatch) │
|
||
│ • 超时时返回部分结果 │
|
||
│ • 熔断器防止级联故障 │
|
||
│ • 消息压缩避免token超限 │
|
||
└───────────────────────────────────────────────────┘
|
||
```
|
||
|
||
### 9.5 范围限定审计
|
||
|
||
```python
|
||
# 支持的范围限定:
|
||
AgentTaskCreate(
|
||
exclude_patterns=["node_modules", "*.min.js"], # 排除模式
|
||
target_files=["src/auth/", "api/views.py"] # 限定文件
|
||
)
|
||
# 减少噪音，聚焦LLM注意力
|
||
```
|
||
|
||
---
|
||
|
||
## 10. 安全与健壮性
|
||
|
||
### 10.1 安全特性
|
||
|
||
| 特性 | 描述 |
|
||
|------|------|
|
||
| **速率限制** | 防止资源耗尽 |
|
||
| **熔断器** | 阻止级联故障 |
|
||
| **输入验证** | 文件路径、大小、扩展名检查 |
|
||
| **资源限制** | 最大发现数(500)、最大文件大小(10MB) |
|
||
| **范围过滤** | 可限定扫描范围 |
|
||
|
||
### 10.2 健壮性特性
|
||
|
||
| 特性 | 描述 |
|
||
|------|------|
|
||
| **指数退避重试** | 处理瞬态故障 |
|
||
| **上下文压缩** | 自动截断长消息历史 |
|
||
| **取消支持** | 可随时中断运行任务 |
|
||
| **检查点** | 长任务恢复点 |
|
||
| **结构化异常** | 分层错误处理 |
|
||
|
||
---
|
||
|
||
## 11. 扩展指南
|
||
|
||
### 11.1 添加新Agent类型
|
||
|
||
```python
|
||
# 1. 创建新Agent类
|
||
class CustomAgent(BaseAgent):
|
||
def __init__(self, config: AgentConfig):
|
||
super().__init__(config)
|
||
self.config.name = "custom"
|
||
self.config.max_iterations = 20
|
||
|
||
async def run(self, input_data: dict) -> AgentResult:
|
||
# 实现ReAct循环
|
||
for iteration in range(self.config.max_iterations):
|
||
response = await self.stream_llm_call(messages)
|
||
parsed = self._parse_response(response)
|
||
|
||
if parsed.action == "finish":
|
||
return AgentResult(data=parsed.data)
|
||
|
||
observation = await self.execute_tool(
|
||
parsed.action,
|
||
parsed.action_input
|
||
)
|
||
messages.append({"role": "user", "content": f"Observation: {observation}"})
|
||
|
||
# 2. 在Orchestrator中注册
|
||
orchestrator.register_agent("custom", CustomAgent(config))
|
||
```
|
||
|
||
### 11.2 添加新工具
|
||
|
||
```python
|
||
# 1. 实现工具接口
|
||
class CustomTool:
|
||
name: str = "custom_tool"
|
||
description: str = "自定义工具描述"
|
||
|
||
async def execute(self, **kwargs) -> dict:
|
||
# 实现工具逻辑
|
||
return {"result": "..."}
|
||
|
||
# 2. 在工具注册表中添加
|
||
TOOL_REGISTRY["custom_tool"] = CustomTool()
|
||
```
|
||
|
||
### 11.3 添加漏洞知识
|
||
|
||
```python
|
||
# 创建新的知识模块: knowledge/vulnerabilities/custom_vuln.py
|
||
|
||
CUSTOM_VULN_KNOWLEDGE = {
|
||
"name": "Custom Vulnerability",
|
||
"description": "...",
|
||
"patterns": [
|
||
r"pattern1",
|
||
r"pattern2"
|
||
],
|
||
"examples": [...],
|
||
"remediation": "..."
|
||
}
|
||
```
|
||
|
||
### 11.4 自定义提示词
|
||
|
||
```python
|
||
# 在system_prompts.py中添加
|
||
CUSTOM_AGENT_PROMPT = """
|
||
你是一个专门的安全分析Agent。
|
||
|
||
职责:
|
||
- ...
|
||
|
||
可用工具:
|
||
- ...
|
||
|
||
输出格式:
|
||
- ...
|
||
"""
|
||
```
|
||
|
||
### 11.5 添加新事件类型
|
||
|
||
```python
|
||
# 在Agent中发射自定义事件
|
||
self.emit_event(
|
||
event_type="custom_event",
|
||
message="自定义消息",
|
||
custom_field="value"
|
||
)
|
||
|
||
# 在前端处理
|
||
switch (event.type) {
|
||
case 'custom_event':
|
||
handleCustomEvent(event);
|
||
break;
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## 附录: 关键文件索引
|
||
|
||
```
|
||
核心Agent逻辑:
|
||
backend/app/services/agent/agents/base.py # BaseAgent基类
|
||
backend/app/services/agent/agents/orchestrator.py # 编排器
|
||
backend/app/services/agent/agents/analysis.py # 分析Agent
|
||
backend/app/services/agent/agents/recon.py # 侦察Agent
|
||
backend/app/services/agent/agents/verification.py # 验证Agent
|
||
|
||
核心基础设施:
|
||
backend/app/services/agent/core/state.py # 状态管理
|
||
backend/app/services/agent/core/context.py # 执行上下文
|
||
backend/app/services/agent/core/circuit_breaker.py # 熔断器
|
||
backend/app/services/agent/core/retry.py # 重试机制
|
||
backend/app/services/agent/core/rate_limiter.py # 速率限制
|
||
backend/app/services/agent/core/validation.py # 输入验证
|
||
backend/app/services/agent/core/errors.py # 错误处理
|
||
|
||
工具:
|
||
backend/app/services/agent/tools/code_analysis_tool.py # 代码分析
|
||
backend/app/services/agent/tools/pattern_tool.py # 模式匹配
|
||
backend/app/services/agent/tools/smart_scan_tool.py # 智能扫描
|
||
backend/app/services/agent/tools/external_tools.py # 外部工具
|
||
backend/app/services/agent/tools/finish_tool.py # 完成工具
|
||
|
||
事件与遥测:
|
||
backend/app/services/agent/event_manager.py # 事件管理
|
||
backend/app/services/agent/telemetry/tracer.py # 分布式追踪
|
||
|
||
配置:
|
||
backend/app/services/agent/config.py # 集中配置
|
||
|
||
API:
|
||
backend/app/api/v1/endpoints/agent_tasks.py # REST端点
|
||
|
||
前端:
|
||
frontend/src/pages/AgentAudit/index.tsx # 主页面
|
||
frontend/src/pages/AgentAudit/hooks/useAgentAuditState.ts # 状态管理
|
||
frontend/src/pages/AgentAudit/hooks/useResilientStream.ts # SSE流
|
||
frontend/src/pages/AgentAudit/types.ts # 类型定义
|
||
frontend/src/pages/AgentAudit/components/* # UI组件
|
||
```
|
||
|
||
---
|
||
|
||
*文档版本: 1.0*
|
||
*最后更新: 2025-12-13*
|