架构总览

相关源文件

本页面内容基于以下源文件生成：

• strix/agents/base_agent.py
• strix/agents/StrixAgent/strix_agent.py
• strix/agents/state.py
• strix/runtime/runtime.py
• strix/tools/registry.py
• strix/llm/llm.py
• strix/tools/agents_graph/agents_graph_actions.py
• strix/llm/config.py
• strix/tools/init.py
• strix/llm/utils.py
• strix/interface/main.py
• Makefile
• pyproject.toml
• containers/Dockerfile
• strix/init.py
• tests/init.py
• tests/conftest.py
• strix/llm/dedupe.py
• strix/llm/init.py
• tests/llm/init.py

Strix 是一个基于 LLM 的自动化安全审计代理框架，采用多 Agent 协作架构，支持任务委托、沙箱隔离和工具调用。该项目的核心设计理念是通过层次化的 Agent 结构实现复杂安全审计任务的分解与执行。

系统架构总览

Strix 采用分层架构设计，从上到下依次为接口层、Agent 层、LLM 集成层、工具层和运行时层。核心组件包括 BaseAgent 基类、AgentState 状态管理、LLM 抽象层、工具注册表以及抽象运行时接口。

正在加载图表渲染器...

架构要点说明：

1. 层次化职责分离：Interface 层负责用户交互，Agent 层负责任务编排，LLM 层负责模型调用，工具层提供具体能力，运行时层管理沙箱环境
2. Agent 继承体系：StrixAgent 继承自 BaseAgent，BaseAgent 通过 AgentMeta 元类实现自动配置加载（strix/agents/base_agent.py:28-46）
3. 状态独立管理：每个 Agent 拥有独立的 AgentState 实例，管理消息历史、迭代计数和执行状态（strix/agents/state.py:12-42）
4. 运行时抽象：通过 AbstractRuntime 接口隔离具体沙箱实现，支持 Docker 等多种后端（strix/runtime/runtime.py:14-33）
5. 工具注册机制：XML Schema 定义工具接口，支持动态技能描述注入（strix/tools/registry.py:47-87）

Agent 核心架构

BaseAgent 基类设计

BaseAgent 是所有 Agent 的抽象基类，通过 AgentMeta 元类实现自动化的资源配置和名称绑定。元类机制在类创建时自动设置 agent_name 和 jinja_env，从 strix_resource_path 加载对应的模板目录（strix/agents/base_agent.py:32-46）。

职责边界：

• ✅ 管理 Agent 生命周期和状态转换
• ✅ 协调 LLM 调用和工具执行
• ✅ 处理沙箱初始化和遥测集成
• ❌ 不直接实现具体业务逻辑（由子类实现）

入口与关键 API：

• __init__(config) — 初始化 Agent 配置、LLM 实例和状态对象（strix/agents/base_agent.py:55-118）
• agent_loop(task) — 主执行循环，处理迭代、工具调用和状态转换（strix/agents/base_agent.py:151-259）
• _process_iteration(tracer) — 处理单次 LLM 迭代，返回是否应结束（strix/agents/base_agent.py:367-413）

关键数据结构：

python
1# Agent 配置结构
2config = {
3    "local_sources": [],           # 本地代码源
4    "max_iterations": 300,         # 最大迭代次数
5    "llm_config_name": "default",  # LLM 配置名称
6    "llm_config": LLMConfig,       # LLM 配置对象
7    "state": AgentState | None     # 可选的状态对象
8}

初始化流程：BaseAgent 在初始化时完成以下关键步骤（strix/agents/base_agent.py:55-118）：

1. 解析配置参数，设置 max_iterations 和 llm_config
2. 创建或复用 AgentState 实例
3. 初始化 LLM 实例并设置 Agent 身份
4. 注册遥测跟踪器，记录 Agent 创建事件
5. 将 Agent 节点添加到全局 Agent 图

AgentState 状态管理

AgentState 是基于 Pydantic BaseModel 的状态容器，管理 Agent 执行过程中的所有可变状态。每个 Agent 拥有独立的状态实例，支持序列化和恢复（strix/agents/state.py:12-42）。

核心状态字段：

状态转换方法（strix/agents/state.py:43-118）：

• increment_iteration() — 递增迭代计数并更新时间戳
• add_message(role, content) — 添加消息到对话历史，支持 thinking_blocks
• set_completed(final_result) — 标记任务完成并存储最终结果
• enter_waiting_state() — 进入等待状态，记录开始时间
• resume_from_waiting() — 从等待状态恢复，重置相关标志
• should_stop() — 检查是否应停止（完成、请求停止或达到最大迭代）

迭代控制逻辑：AgentState 提供多层迭代阈值检查（strix/agents/state.py:113-118）：

python
1def has_reached_max_iterations(self) -> bool:
2    return self.iteration >= self.max_iterations
3
4def is_approaching_max_iterations(self, threshold: float = 0.85) -> bool:
5    return self.iteration >= int(self.max_iterations * threshold)

StrixAgent 具体实现

StrixAgent 是 BaseAgent 的具体实现，作为安全审计任务的入口 Agent。它在初始化时根据是否为根 Agent 自动配置默认技能（strix/agents/StrixAgent/strix_agent.py:7-19）。

入口方法 execute_scan（strix/agents/StrixAgent/strix_agent.py:21-89）：

该方法接收扫描配置，解析目标类型（仓库、本地代码、URL、IP 地址），构建任务描述并启动 Agent 循环：

python
1async def execute_scan(self, scan_config: dict[str, Any]) -> dict[str, Any]:
2    targets = scan_config.get("targets", [])
3    
4    for target in targets:
5        target_type = target["type"]
6        if target_type == "repository":
7            repositories.append({...})
8        elif target_type == "local_code":
9            local_code.append({...})
10        elif target_type == "web_application":
11            urls.append(details["target_url"])
12    
13    return await self.agent_loop(task=task_description)

目标类型处理：

Agent 执行循环

主循环 agent_loop 实现

agent_loop 是 Agent 的核心执行入口，实现了一个异步事件循环，持续处理 LLM 响应和工具调用直到任务完成或达到停止条件（strix/agents/base_agent.py:151-259）。

循环控制流程：

正在加载图表渲染器...

停止条件检查（strix/agents/base_agent.py:173-181）：

循环在以下情况下暂停或终止：

• state.should_stop() 返回 True（完成、请求停止或达到最大迭代）
• state.is_waiting_for_input() 为 True（等待用户输入）
• state.llm_failed 为 True（LLM 请求失败）

迭代警告机制：当迭代次数接近阈值时，系统自动注入警告消息（strix/agents/base_agent.py:185-210）：

• 85% 阈值：发送优先级警告，提醒尽快完成任务
• 剩余 3 次迭代：发送关键警告，强制要求调用结束工具

迭代处理与工具执行

_process_iteration 方法处理单次 LLM 迭代，包括流式响应处理、消息记录和工具调用执行（strix/agents/base_agent.py:367-413）。

LLM 响应处理流程：

1. 调用 llm.generate() 获取流式响应
2. 实时更新遥测跟踪器的流式内容
3. 检查空响应并注入纠正消息
4. 提取 tool_invocations 并执行

空响应纠正逻辑（strix/agents/base_agent.py:380-392）：

python
1if not content_stripped:
2    corrective_message = (
3        "You MUST NOT respond with empty messages. "
4        "If you currently have nothing to do or say, use an appropriate tool instead:\n"
5        "- Use agents_graph_actions.wait_for_message to wait for messages\n"
6        "- Use agents_graph_actions.agent_finish if you are a sub-agent\n"
7        "- Use finish_actions.finish_scan if you are the root/main agent"
8    )
9    self.state.add_message("user", corrective_message)
10    return False

工具执行流程（strix/agents/base_agent.py:415-428）：

_execute_actions 方法将工具调用委托给 process_tool_invocations 函数，该函数负责解析工具名称、验证参数并执行对应的工具函数。执行结果作为观察添加到状态中。

等待状态与错误处理

等待状态管理：Agent 支持进入等待状态以接收用户输入或其他 Agent 消息（strix/agents/base_agent.py:286-328）。

_enter_waiting_state 方法根据不同场景设置状态和遥测：

• text_response=True：LLM 返回纯文本响应，等待后续指令
• task_completed=True：任务完成，等待后续任务
• error_occurred=True：发生错误，等待新指令
• was_cancelled=True：执行被取消，等待恢复

等待超时机制（strix/agents/state.py:119-135）：

python
1def has_waiting_timeout(self) -> bool:
2    if self.waiting_timeout == 0:
3        return False  # 交互模式根 Agent 永不超时
4    
5    elapsed = (datetime.now(UTC) - self.waiting_start_time).total_seconds()
6    return elapsed > self.waiting_timeout

异常处理策略（strix/agents/base_agent.py:231-258）：

多 Agent 协作机制

Agent 图结构管理

Strix 维护一个全局的 Agent 图数据结构，记录所有 Agent 节点、边关系和消息队列。该结构由 agents_graph_actions 模块管理（strix/tools/agents_graph/agents_graph_actions.py:24-102）。

全局数据结构：

python
1_agent_graph = {
2    "nodes": {},  # agent_id -> node_info
3    "edges": []   # [{"from": parent_id, "to": child_id, "type": "delegation"}]
4}
5_agent_instances = {}    # agent_id -> agent_object
6_agent_states = {}       # agent_id -> state_object
7_agent_messages = {}     # agent_id -> message_queue
8_root_agent_id = None    # 根 Agent ID

Agent 注册流程（strix/agents/base_agent.py:119-150）：

每个 Agent 在初始化时调用 _add_to_agents_graph 方法：

1. 创建节点信息字典，包含 id、name、task、status 等字段
2. 将节点添加到 _agent_graph["nodes"]
3. 将 Agent 实例和状态添加到全局注册表
4. 如果有父 Agent，创建委托边
5. 初始化消息队列
6. 如果是首个根 Agent，设置 _root_agent_id

子 Agent 创建与委托

子 Agent 通过 _run_agent_in_thread 函数在独立线程中运行，继承父 Agent 的上下文消息但保持独立的对话历史（strix/tools/agents_graph/agents_graph_actions.py:24-102）。

上下文继承机制：

python
1if inherited_messages:
2    state.add_message("user", "<inherited_context_from_parent>")
3    for msg in inherited_messages:
4        state.add_message(msg["role"], msg["content"])
5    state.add_message("user", "</inherited_context_from_parent>")

身份隔离指令：系统通过 XML 指令明确告知子 Agent 其身份边界（strix/tools/agents_graph/agents_graph_actions.py:43-68）：

xml
1<agent_delegation>
2    <identity>
3        ⚠️ You are NOT your parent agent. You are a NEW, SEPARATE sub-agent.
4        Your Info: {state.agent_name} ({state.agent_id})
5        Parent Info: {parent_name} ({state.parent_id})
6    </identity>
7    <instructions>
8        - Inherited context is for BACKGROUND ONLY
9        - Maintain strict self-identity: never speak as or for your parent
10        - Use agent_finish when complete to report back to parent
11    </instructions>
12</agent_delegation>

线程执行模型：子 Agent 在新的事件循环中运行（strix/tools/agents_graph/agents_graph_actions.py:76-84）：

python
1loop = asyncio.new_event_loop()
2asyncio.set_event_loop(loop)
3try:
4    result = loop.run_until_complete(agent.agent_loop(state.task))
5finally:
6    loop.close()

Agent 间通信

Agent 图可视化：view_agent_graph 工具提供 Agent 树状结构的可视化视图（strix/tools/agents_graph/agents_graph_actions.py:106-175）。

树状结构构建：

python
1def _build_tree(agent_id: str, depth: int = 0) -> None:
2    node = _agent_graph["nodes"][agent_id]
3    indent = "  " * depth
4    you_indicator = " ← This is you" if agent_id == agent_state.agent_id else ""
5    
6    structure_lines.append(f"{indent}* {node['name']} ({agent_id}){you_indicator}")
7    structure_lines.append(f"{indent}  Task: {node['task']}")
8    structure_lines.append(f"{indent}  Status: {node['status']}")
9    
10    # 递归构建子节点
11    for child_id in children:
12        _build_tree(child_id, depth + 2)

状态统计：该工具还提供 Agent 状态的聚合统计（strix/tools/agents_graph/agents_graph_actions.py:147-165）：

LLM 集成层

LLM 响应模型

LLM 集成层定义了标准的响应模型和错误类型（strix/llm/llm.py:29-58）。

LLMResponse 数据类：

python
1@dataclass
2class LLMResponse:
3    content: str
4    tool_invocations: list[dict[str, Any]] | None = None
5    thinking_blocks: list[dict[str, Any]] | None = None

RequestStats 统计类：跟踪 LLM 调用的资源消耗（strix/llm/llm.py:44-58）：

python
1@dataclass
2class RequestStats:
3    input_tokens: int = 0
4    output_tokens: int = 0
5    cached_tokens: int = 0
6    cost: float = 0.0
7    requests: int = 0

请求错误处理

LLMRequestFailedError 异常：封装 LLM 请求失败的详细信息（strix/llm/llm.py:29-33）：

python
1class LLMRequestFailedError(Exception):
2    def __init__(self, message: str, details: str | None = None):
3        super().__init__(message)
4        self.message = message
5        self.details = details

在 Agent 循环中，该异常被捕获并调用 _handle_llm_error 方法处理（strix/agents/base_agent.py:244-248）。

工具注册与动态 Schema

工具注册表：registry.py 负责加载 XML 格式的工具 Schema 并处理动态内容（strix/tools/registry.py:47-87）。

动态技能描述注入（strix/tools/registry.py:30-43）：

python
1def _process_dynamic_content(content: str) -> str:
2    if "{{DYNAMIC_SKILLS_DESCRIPTION}}" in content:
3        try:
4            from strix.skills import generate_skills_description
5            skills_description = generate_skills_description()
6            content = content.replace("{{DYNAMIC_SKILLS_DESCRIPTION}}", skills_description)
7        except ImportError:
8            content = content.replace(
9                "{{DYNAMIC_SKILLS_DESCRIPTION}}",
10                "List of skills to load for this agent (max 5). Skill discovery failed."
11            )
12    return content

XML Schema 解析：_load_xml_schema 函数解析工具定义文件，提取工具名称和定义（strix/tools/registry.py:55-87）。

运行时与沙箱环境

抽象运行时接口

AbstractRuntime 定义了沙箱管理的抽象接口，支持多种后端实现（strix/runtime/runtime.py:14-33）。

SandboxInfo 类型定义（strix/runtime/runtime.py:5-11）：

python
1class SandboxInfo(TypedDict):
2    workspace_id: str
3    api_url: str
4    auth_token: str | None
5    tool_server_port: int
6    caido_port: int
7    agent_id: str

核心抽象方法：

沙箱初始化流程

Agent 在首次执行前初始化沙箱环境（strix/agents/base_agent.py:330-365）。

初始化逻辑：

python
1async def _initialize_sandbox_and_state(self, task: str) -> None:
2    sandbox_mode = os.getenv("STRIX_SANDBOX_MODE", "false").lower() == "true"
3    
4    if not sandbox_mode and self.state.sandbox_id is None:
5        runtime = get_runtime()
6        sandbox_info = await runtime.create_sandbox(
7            self.state.agent_id,
8            self.state.sandbox_token,
9            self.local_sources
10        )
11        self.state.sandbox_id = sandbox_info["workspace_id"]
12        self.state.sandbox_token = sandbox_info["auth_token"]
13        
14        # 设置 Caido 代理 URL
15        caido_port = sandbox_info.get("caido_port")
16        if caido_port:
17            tracer.caido_url = f"localhost:{caido_port}"

错误处理：沙箱初始化失败时，通过 PostHog 报告错误并抛出 SandboxInitializationError（strix/agents/base_agent.py:356-360）。

核心数据流

以下序列图展示了从用户请求到 Agent 执行的完整数据流：

正在加载图表渲染器...

数据流要点：

1. 请求入口：用户请求通过 Interface 层进入，调用 StrixAgent 的 execute_scan 方法
2. 目标解析：StrixAgent 解析扫描配置中的目标类型，构建任务描述
3. 沙箱初始化：BaseAgent 在首次迭代前调用 Runtime 创建隔离环境
4. 迭代循环：持续调用 LLM 生成响应，执行工具调用，直到满足停止条件
5. 状态管理：AgentState 贯穿整个生命周期，记录消息、动作和观察

核心设计决策与取舍

1. 元类驱动的资源配置

决策：使用 AgentMeta 元类自动设置 Agent 名称和 Jinja 环境（strix/agents/base_agent.py:28-46）。

理由：

• 避免每个子类重复编写资源配置代码
• 约定优于配置：模板目录路径基于类名自动推导
• 支持热加载：类定义时即可完成资源绑定

限制：模板目录必须遵循 {resource_path}/agents/{ClassName} 命名约定。

2. 异步迭代循环模型

决策：采用异步事件循环而非有限状态机（strix/agents/base_agent.py:151-259）。

理由：

• LLM 调用天然适合异步模型
• 支持流式响应处理和实时遥测
• 简化并发子 Agent 的管理

限制：复杂状态转换逻辑分散在多个条件判断中，可读性略低。

3. 全局 Agent 图结构

决策：使用模块级全局变量维护 Agent 关系图（strix/agents/base_agent.py:119-150）。

理由：

• 简化跨 Agent 通信：任何 Agent 都可直接访问图结构
• 支持动态发现：子 Agent 可查询父 Agent 信息
• 遥测集成：全局视图便于跟踪整个任务树

限制：单进程假设，不支持分布式部署。

4. 线程隔离的子 Agent 执行

决策：子 Agent 在独立线程和事件循环中运行（strix/tools/agents_graph/agents_graph_actions.py:76-84）。

理由：

• 隔离异步上下文：避免事件循环冲突
• 简化错误隔离：子 Agent 崩溃不影响父 Agent
• 支持并行执行：多个子 Agent 可同时运行

限制：线程间通信需要通过全局数据结构，增加同步复杂度。

5. XML 格式的工具 Schema

决策：使用 XML 而非 JSON 定义工具接口（strix/tools/registry.py:47-87）。

理由：

• 更好的可读性：嵌套结构清晰
• 支持动态占位符：{{DYNAMIC_SKILLS_DESCRIPTION}}
• 与 LLM 提示词格式一致

限制：解析复杂度高于 JSON，需要手动处理标签匹配。

6. 迭代警告机制

决策：在接近最大迭代时注入警告消息（strix/agents/base_agent.py:185-210）。

理由：

• 防止无限循环：强制 Agent 在限制内完成
• 渐进式提醒：85% 和 97% 两级阈值
• 保持上下文：警告作为用户消息添加，不破坏对话流

限制：依赖 LLM 正确理解警告并采取行动。

7. 抽象运行时接口

决策：定义 AbstractRuntime 抽象类而非直接依赖 Docker（strix/runtime/runtime.py:14-33）。

理由：

• 后端可替换：支持 Docker、Kubernetes、Mock 等
• 便于测试：可注入 Mock 实现
• 清晰的契约：明确沙箱生命周期管理职责

限制：增加抽象层，可能隐藏实现细节。

技术选型

模块依赖关系

正在加载图表渲染器...

依赖方向说明：

1. 接口层 → 领域层：Interface 依赖 Agent 模块启动任务
2. 领域层 → 基础设施层：Agent 依赖 Runtime、LLM、Telemetry
3. 领域层 → 工具层：Agent 调用工具注册表和 Agent 图操作
4. 基础设施层内部：Runtime 和 LLM 可能依赖 Telemetry

关键配置与启动流程

环境变量配置

Agent 配置结构

python
1agent_config = {
2    "local_sources": [
3        {"path": "/local/code", "workspace_path": "/workspace/code"}
4    ],
5    "max_iterations": 300,
6    "llm_config_name": "default",
7    "llm_config": LLMConfig(
8        skills=["root_agent"],
9        interactive=False
10    ),
11    "state": None  # 可选，传入已有的 AgentState
12}

启动流程

1. 创建 StrixAgent 实例：传入配置字典
2. 调用 execute_scan：传入扫描配置
3. Agent 循环执行：自动管理沙箱、LLM 调用和工具执行
4. 返回最终结果：包含成功状态和审计发现

扫描配置示例：

python
1scan_config = {
2    "targets": [
3        {
4            "type": "repository",
5            "details": {
6                "target_repo": "https://github.com/example/repo",
7                "cloned_repo_path": "/tmp/repo",
8                "workspace_subdir": "repo"
9            }
10        }
11    ],
12    "user_instructions": "重点关注认证和授权漏洞"
13}