AgentWorld：为强智能体构建文件系统原生工作流的底层平台

1. 项目概述：AgentWorld，一个为强智能体构建原生工作流的底层平台

如果你最近在尝试构建一个由多个AI智能体协作的自动化系统，比如一个能自动完成文献调研、代码编写、实验执行和论文撰写的“AI研究员”，你可能会发现，现有的很多框架更像是“胶水”——它们帮你调用模型API，但把真正棘手的问题留给了你：如何管理智能体之间复杂的对话？如何持久化运行状态以便随时中断和恢复？如何确保每个智能体都能访问正确的工具和上下文？如何验证最终产出的文件是否符合要求？

AgentWorld正是为了解决这些问题而生的。它不是另一个高级的“无代码”应用构建器，而是一个底层Python包，专注于为“强智能体”构建文件系统原生的工作流和组织架构。这里的“强智能体”指的是那些能够真正运行会话、操作工作目录、调用有副作用的工具，并进行长时间状态化执行的AI系统，比如Claude Code。AgentWorld提供了一套可复用的基础组件——控制器、操作器、工作空间、阶段、工件、审批门和技能市场——让你可以在此基础上构建像AutoR这样的自动化研究系统，或者任何需要可靠、可追踪、可恢复的多智能体应用。

简单来说，当其他框架在思考“模型能调用什么”时，AgentWorld的起点是“一个强智能体实际能运行什么”。这种思路的转变带来了架构上的根本不同：控制器不再是可选的粘合代码，而是与具体AI提供商交互的边界；操作器不仅仅是提示词包装器，而是一个持久的执行单元；技能不是营销标签，而是可以附加到操作器上的、可复用的执行指导模块。

2. 核心架构与设计哲学拆解

2.1 为什么是“文件系统原生”？

在接触AgentWorld时，你首先会注意到它对文件系统的重度依赖。这并非偶然，而是其设计哲学的核心体现。传统的多智能体框架往往将状态保存在内存或临时数据库中，一旦进程崩溃或需要横向扩展，状态管理就变得异常复杂。

AgentWorld选择将每一次智能体协作的运行（称为一个“Run”）都锚定在一个具体的目录结构中。这个run_root目录包含了从目标描述、运行配置、阶段清单、工件索引到完整工作空间的一切。这样做有几个关键优势：

1. 可恢复性：运行状态被持久化到磁盘。如果系统因任何原因中断（如网络故障、手动停止），你可以从上次成功的检查点（checkpoint）恢复，无需从头开始。这对于可能运行数小时的研究工作流至关重要。
2. 可审计与可调试：所有中间产物——智能体的原始输出、工具调用记录、生成的代码、实验数据、图表——都按结构存放。你可以像检查一个人类研究员的工作日志一样，逐层深入查看AI的工作过程，定位问题所在。
3. 强智能体的“自然栖息地”：像Claude Code这样的强智能体，其核心能力就是在一个真实的文件系统环境中读写文件、执行命令。AgentWorld的工作空间（workspace/）正是为它们提供的沙箱。智能体可以直接在workspace/code/下编写Python脚本，在workspace/data/下处理数据，最终在workspace/artifacts/下生成报告PDF。整个流程与人类工作流高度同构。

2.2 核心分层：从控制器到应用

AgentWorld的架构是分层的，当前代码库主要实现了最基础的“平台基础层”。理解这些层次有助于你定位自己的开发工作应该在哪里进行。

控制器：这是与特定AI提供商（如Claude Code、Codex）交互的边界。它负责会话生命周期管理、工具策略映射以及原始事件流的解析。例如，ClaudeCodeController会封装对本地claudeCLI的调用，并将Claude Code返回的流式输出解析成AgentWorld内部统一的ControllerEvent。

操作器：它位于控制器之上，提供统一的请求/结果合约。操作器接收来自运行时的标准化请求（包含目标、角色、技能、状态等），组装成适合控制器的提示，并将控制器返回的事件归一化为消息、工具结果、工件引用等结构化输出。它是将平台逻辑与提供商细节解耦的关键。

A2A协议：即“智能体到智能体”协议。它定义了智能体间通信的结构化格式，包括消息、工具输出、交接文档和工件引用。这避免了传统多智能体系统中常见的、脆弱的“提示词胶水”问题，使得智能体间的协作意图更加明确和可追踪。

运行时：这是执行引擎，负责调度图节点、合并状态、管理检查点、处理中断和恢复，并持久化可追踪的执行状态。它确保整个工作流是状态化的、可恢复的。

技能：这是AgentWorld一个非常实用的设计。技能是可复用的执行指导模块，以文件夹形式存在（如skills/research-paper-search/），包含SKILL.md指导文档，以及可选的参考资料、脚本和资产。你可以在图的不同节点上为操作器附加不同的技能集，从而让同一个Claude Code实例在“规划”节点上具备文献检索能力，在“评审”节点上具备结果审计能力，而无需修改底层控制器的配置。

3. 核心组件深度解析与实操要点

3.1 工作空间与运行清单：状态的基石

创建一个运行工作空间是你使用AgentWorld的第一步。这个操作不仅创建目录，更初始化了整个运行的状态管理框架。

from pathlib import Path from agentworld import create_run_workspace workspace = create_run_workspace( runs_dir=Path("./my_research_runs"), # 所有运行的根目录 run_id="digits_classification_20250423", # 唯一运行ID goal="构建并评估一个基于scikit-learn手写数字数据集的分类模型，比较多种基线算法并生成研究报告。", config={ "workflow": "auto-research", "approval_mode": "validation-only", "permission_mode": "bypassPermissions" }, )

运行后，你的./my_research_runs/digits_classification_20250423/目录结构将如下所示。理解每个文件和目录的用途，是后续调试和定制的关键：

digits_classification_20250423/ ├── goal.md # 运行目标的纯文本描述 ├── memory.md # 跨阶段批准的“记忆”，是智能体间传递上下文的核心 ├── run_config.json # 运行的配置（目标、模式等） ├── run_manifest.json # **核心**：运行清单，记录所有阶段的状态、结果和元数据 ├── artifact_index.json # 工件索引，记录所有产出文件的路径、类型和模式推断 ├── logs.txt # 聚合的运行日志 ├── events.jsonl # 原始的、按行存储的运行时事件流 ├── logs_raw/ # 每个控制器会话的原始日志 ├── prompt_cache/ # 提示词缓存，用于加速和调试 ├── operator_state/ # 每个操作器节点的持久化状态 ├── stages/ # 每个阶段的工作目录，包含草稿、最终版等 ├── handoffs/ # 阶段间的结构化交接文档 └── workspace/ # 智能体实际工作的沙箱 ├── literature/ # 文献资料 ├── code/ # 生成的代码 ├── data/ # 数据集或处理后的数据 ├── results/ # 实验结果（JSON、NPZ等） ├── figures/ # 生成的图表 ├── writing/ # 报告文稿（LaTeX, Bib等） ├── notes/ # 临时笔记 ├── reviews/ # 评审意见 ├── artifacts/ # 最终产出物（PDF、构建日志等） ├── bootstrap/ # 运行初始化脚本 └── profile/ # 性能分析数据

实操心得：run_manifest.json是你的控制面板在运行过程中，定期检查run_manifest.json是了解进度的最佳方式。它会实时更新每个阶段（如literature_review,hypothesis_generation,experiment_execution）的状态（pending,running,completed,failed）、批准状态、开始/结束时间以及结果摘要。如果运行卡住或失败，这是你第一个应该查看的文件。

3.2 图运行时与技能注入：构建动态工作流

AgentWorld的图运行时允许你定义由操作器节点组成的工作流。每个节点可以绑定不同的技能，实现动态的能力组合。

下面是一个构建“规划-编码-评审”三节点研究流程的示例，并展示了如何为不同节点注入不同的技能集：

from agentworld import AgentGraph, DefaultOperator from agentworld.controller.claude_code import ClaudeCodeController from agentworld.controller.base import ControllerEvent import asyncio # 1. 定义操作器（这里使用一个模拟控制器作为示例） def mock_planner_script(request): """模拟规划器控制器：接收请求，返回结构化事件。""" # request中包含了运行时注入的skills信息 active_skills = request.get("skills", []) skill_context = f"激活的技能: {', '.join(active_skills)}" if active_skills else "未激活特定技能。" return [ ControllerEvent(kind="message_completed", payload={ "text": f"基于目标'{request.get('objective')}'，我已制定研究计划。{skill_context}" }), ControllerEvent(kind="tool_call", payload={ "name": "save_plan", "arguments": {"path": "workspace/literature/research_plan.md"}, "call_id": "call_001" }), ControllerEvent(kind="tool_result", payload={ "call_id": "call_001", "result": {"success": True, "path": "workspace/literature/research_plan.md"} }), ControllerEvent(kind="completed", payload={ "state_patch": {"plan_ready": True, "next_step": "coding"} }), ] def mock_coder_script(request): """模拟编码器控制器。""" return [ ControllerEvent(kind="message_completed", payload={ "text": f"已根据计划编写实现代码。" }), ControllerEvent(kind="completed", payload={ "state_patch": {"code_written": True, "files": ["workspace/code/train.py"]} }), ] def mock_reviewer_script(request): """模拟评审器控制器。""" return [ ControllerEvent(kind="message_completed", payload={ "text": f"已完成代码和结果的审计。" }), ControllerEvent(kind="completed", payload={ "state_patch": {"review_passed": True} }), ] # 2. 创建图并添加操作器 graph = AgentGraph(name="research-pipeline") graph.add_operator("planner", DefaultOperator("planner", mock_planner_script)) graph.add_operator("coder", DefaultOperator("coder", mock_coder_script)) graph.add_operator("reviewer", DefaultOperator("reviewer", mock_reviewer_script)) # 3. 添加节点，并为每个节点指定不同的技能 # 规划节点：需要文献检索和实验规划技能 graph.add_node( "plan_node", operator="planner", objective="进行文献调研并制定详细的实验计划", skills=["research-paper-search", "experiment-planning"], # 技能注入点 ) # 编码节点：可能不需要额外的领域技能，专注于实现 graph.add_node( "code_node", operator="coder", objective="根据计划编写可执行的模型训练与评估代码", skills=[], # 可以留空或注入代码规范类技能 ) # 评审节点：需要引用审计和结果审计技能 graph.add_node( "review_node", operator="reviewer", objective="审计实验代码、结果的有效性及报告中的引用", skills=["citation-audit", "result-audit"], ) # 4. 定义节点间的边（执行顺序） graph.add_edge("plan_node", "code_node") graph.add_edge("code_node", "review_node") # 5. 编译并执行图 async def run_graph(): compiled_graph = graph.compile() # 初始状态可以传递任务描述 initial_state = {"task": "构建一个手写数字分类器的研究原型"} result = await compiled_graph.ainvoke(initial_state) print("最终状态:", result.state) print("生成的工件:", result.artifacts) # 运行 asyncio.run(run_graph())

在这个例子中，skills参数是关键。运行时会将指定的技能文件夹（如skills/research-paper-search/）中的SKILL.md内容读取并注入到发送给操作器的请求中。这意味着，即使planner和reviewer使用同一个ClaudeCodeController，它们在执行时收到的指令和上下文也会因技能的不同而不同，从而实现了基于节点的能力定制。

注意事项：技能加载机制技能是通过文件系统路径加载的。确保你的技能文件夹结构正确，并且SKILL.md的YAML头信息（如name,description）完整。运行时只会注入SKILL.md的主体内容作为指导，不会自动执行scripts/下的脚本，这些脚本需要你在技能描述中明确指示智能体去调用。

3.3 自动化研究用例深度实操

AgentWorld内置了一个完整的自动化研究（Auto-Research）工作流示例，这是展示其能力的最佳场景。这个工作流模拟了一个完整的科研过程，包含八个阶段。我们来看如何运行并验证一个真实的案例。

第一步：环境准备与运行确保你已安装AgentWorld并配置好Claude Code CLI（因为示例默认使用它）。然后运行一个完整的自动化研究任务：

# 切换到项目目录 cd AgentWorld # 运行一个要求执行真实实验的任务 python examples/auto-research/run.py \ --runs-dir /tmp/agentworld-runs \ --approval-mode validation-only \ --permission-mode bypassPermissions \ --max-attempts 2 \ --timeout 7200 \ "在scikit-learn的digits数据集上构建并评估手写数字分类模型。训练SVM-RBF、随机森林、kNN、逻辑回归和决策树基线。实验阶段必须实际执行Python训练脚本，并产生真实的交叉验证结果、留出测试结果、混淆矩阵、假设检验结论、图表以及论文风格的报告。不得使用预测或仅文献中的结果替代实际执行。"

关键参数解析：

• --approval-mode validation-only: 使用基于验证的自动批准，而非人工交互批准。适合自动化流水线。
• --permission-mode bypassPermissions: 允许智能体在工作空间内执行命令（如python workspace/code/train.py）。这是实验能真实执行的关键。如果设为更安全的editOnly，智能体将无法运行代码，实验阶段会失败。
• --max-attempts 2: 每个阶段失败后最多重试2次。
• --timeout 7200: 整个运行的总超时时间为2小时。

第二步：验证运行结果运行完成后，不能只看run_manifest.json显示completed就认为成功。必须检查关键工件是否生成，以及实验结果是否来自真实执行。

# 找到最新的运行目录 RUN_ROOT=$(ls -td /tmp/agentworld-runs/* | head -1) # 使用Python脚本进行系统化验证 python - <<PY import json from pathlib import Path root = Path("$RUN_ROOT") print(f"检查运行目录: {root}") # 1. 检查运行清单状态 manifest_path = root / "run_manifest.json" if not manifest_path.exists(): print("错误: 未找到 run_manifest.json") exit(1) manifest = json.loads(manifest_path.read_text()) print(f"运行状态: {manifest.get('run_status')}") print(f"阶段批准情况: {sum(1 for s in manifest['stages'] if s.get('approved'))}/{len(manifest['stages'])}") # 2. 检查核心结果文件 results_path = root / "workspace/results/results.json" if not results_path.exists(): print("错误: 未找到 results.json") exit(1) results = json.loads(results_path.read_text()) # 关键验证点 execution_blocked = results.get("execution_blocker", False) experiments_executed = results.get("experiments_executed", False) exit_code = results.get("exit_code") print(f"执行是否被阻止: {execution_blocked}") print(f"实验是否已执行: {experiments_executed}") print(f"实验脚本退出码: {exit_code}") if execution_blocked: print("致命错误: 实验执行被权限策略阻止。") exit(1) if not experiments_executed: print("致命错误: 实验未实际执行，结果可能来自预测。") exit(1) if exit_code not in [0, None]: # None可能表示非脚本执行 print(f"警告: 实验脚本可能非正常退出 (exit_code={exit_code})。") # 3. 检查必需工件是否存在 required_artifacts = [ "workspace/results/cv_results.json", "workspace/results/test_results.json", "workspace/results/hypothesis_verdicts.json", "workspace/figures/accuracy_comparison.png", "workspace/figures/confusion_matrices.png", "workspace/writing/main.tex", "workspace/artifacts/paper.pdf", ] missing = [a for a in required_artifacts if not (root / a).exists()] if missing: print(f"警告: 缺失以下工件: {missing}") else: print("成功: 所有核心工件均已生成。") # 4. 检查实验结果质量（示例：查看交叉验证精度） cv_results_path = root / "workspace/results/cv_results.json" if cv_results_path.exists(): cv_data = json.loads(cv_results_path.read_text()) print("\n模型交叉验证平均准确率:") for model, data in cv_data.items(): if isinstance(data, dict): acc = data.get('mean_cv_accuracy') if acc is not None: print(f" {model}: {acc:.4f}") PY

这个验证脚本是必须的。它检查了三个层面：

1. 流程完整性：运行是否完成，所有阶段是否获批。
2. 执行真实性：实验是否真的被执行了（而非被阻止或跳过）。
3. 产出完备性：关键的结果文件、图表和报告是否生成。

避坑指南：权限模式的选择bypassPermissions模式赋予了智能体很大的权力，因为它允许执行任意命令。仅在受信任的环境或沙箱中使用。对于生产环境，你应该：

1. 使用editOnly模式，并让智能体生成可复现的脚本。
2. 在后续的“审批门”或自定义阶段中，由另一个受信任的进程（或经过人工审核后）来执行这些脚本。
3. 或者，实现一个更细粒度的ToolPolicy，只允许运行白名单内的命令（如python、pip install等）。AgentWorld的控制器架构支持这种定制。

3.4 技能市场的创建与使用

技能是AgentWorld实现领域知识复用的核心。让我们深入看看如何创建和使用一个自定义技能。

创建一个新技能：>--- name:>graph.add_node( "data_prep_node", operator="data_specialist", objective="对提供的数据集进行清洗和质量检查，为建模做准备", skills=["data-cleaning"], # 注入自定义技能 )

当运行时执行到data_prep_node时，它会读取skills/data-cleaning/SKILL.md的内容，并将其作为额外的上下文指令插入到发送给data_specialist操作器的请求中。这样，即使操作器本身没有内置数据清洗逻辑，也能通过技能获得专业的执行指导。

经验技巧：让技能更有效

• 具体化：技能描述越具体、越像给新手的 checklist，智能体执行得越好。避免模糊的“请检查数据质量”。
• 结构化输出：在技能中明确要求智能体更新状态（state_patch）和生成特定路径的工件。这便于下游节点依赖这些产出。
• 提供可复用资产：scripts/和references/里的内容是给智能体“看”和“用”的。你可以在SKILL.md中写道：“你可以参考references/common_issues.md中的模式，或者直接调用scripts/validate_schema.py来辅助验证。”

4. 高级特性与内部机制剖析

4.1 A2A协议：告别“提示词胶水”

在多智能体系统中，一个常见的反模式是让智能体A生成一段文本，然后手动（或通过脆弱的字符串解析）将其塞进给智能体B的提示词里。AgentWorld的A2A协议旨在用结构化的消息取代这种“胶水”。

协议定义了几种核心的交互类型：

• 消息：智能体间的纯文本通信。
• 工具调用与结果：智能体调用工具（如读写文件、执行命令）的标准化请求和响应。
• 交接文档：一个阶段完成后，传递给下一个阶段的正式、结构化的上下文摘要。
• 工件引用：对生成的文件（如图表、数据、报告）的明确引用，包含路径、类型和描述。

在运行时，操作器负责将控制器的原始输出（如Claude Code的tool_use事件）归一化为这些协议对象。图运行时则负责将这些对象路由到正确的下游节点，并合并到全局状态中。这使得智能体间的协作意图清晰、可追踪，并且极大地降低了集成复杂度。

4.2 检查点与恢复：实现长时程可靠性

对于可能运行数小时甚至数天的研究流程，容错性至关重要。AgentWorld的运行时内置了检查点机制。

1. 状态快照：在每个图节点执行完成后，运行时会将其产生的state_patch与全局状态合并，并将完整状态持久化到operator_state/目录下。
2. 清单更新：run_manifest.json会实时记录每个阶段和节点的状态（completed,failed）。
3. 恢复执行：如果运行因错误或手动中断而停止，你可以使用--resume-run <run_root>参数来恢复。运行时会读取清单和状态，跳过已完成的阶段，并从第一个失败或未开始的节点继续执行。
4. 回滚：更高级的rollback_to_stage功能允许你回滚到某个特定阶段，丢弃其之后的所有结果，然后重新执行。这在发现早期阶段有根本性错误时非常有用。

# 恢复一个中断的运行 python examples/auto-research/run.py \ --resume-run /tmp/agentworld-runs/digits_classification_20250423 \ --approval-mode validation-only

实现细节：状态合并策略运行时使用“补丁合并”策略。每个节点返回的state_patch是一个字典，运行时会将其浅合并到全局状态中。这意味着后执行的节点可以覆盖先前节点的状态值。对于复杂的嵌套状态，你需要设计好键名以避免冲突。常见的模式是使用命名空间，如state_patch: {"literature": {"key_papers": [...]}, "experiment": {"models_trained": [...]}}。

4.3 工件验证与模式推断

智能体生成的文件（工件）可能格式不一、内容良莠不齐。AgentWorld提供了ArtifactRequirement和scan_artifacts工具来应对。

• 工件需求：你可以在阶段定义中指定必须生成的工件，例如，experiment阶段必须产出results.json和confusion_matrix.png。
• 自动扫描与索引：scan_artifacts函数会遍历工作空间，根据文件扩展名和内容（尝试）推断其模式（如JSON、图像、文本、LaTeX），并生成artifact_index.json。这个索引文件记录了每个工件的路径、推断的类型和大小，方便后续阶段或审批门进行验证。
• 模式验证：你可以编写简单的验证逻辑，检查results.json是否包含必需的键（如accuracy），或者confusion_matrix.png文件大小是否合理（避免空图）。

这种机制将“是否有文件”和“文件是否正确”的检查从模糊的提示词要求，变成了系统中可编程、可自动执行的约束。

5. 常见问题、排查技巧与扩展方向

5.1 常见问题速查表

5.2 性能优化与调试技巧

1. 利用提示词缓存：AgentWorld会缓存编译后的提示词到prompt_cache/。在开发阶段，如果你频繁修改技能或节点目标，可以临时清空此目录以强制刷新缓存。在生产环境中，缓存能显著提升性能。
2. 精细化日志：默认日志可能不够详细。你可以通过环境变量或配置增加运行时和控制器日志的详细程度，特别是在调试协议转换或技能注入问题时。
3. 分阶段测试：不要一开始就运行完整的八阶段研究流程。先构建一个最小的两节点图（如planner->reviewer），确保基础通信和状态流转正常，再逐步添加复杂阶段和技能。
4. 模拟控制器先行：在集成真实的ClaudeCodeController之前，先用StaticController（返回预定义事件的模拟控制器）来测试你的图逻辑、技能注入和状态合并。这能快速验证业务逻辑，无需消耗API调用。

5.3 扩展方向：构建你自己的智能体应用

AgentWorld的基础设施是通用的。你可以超越自动化研究，构建自己的智能体应用：

1. 自定义工作流：定义你自己的StageSpec序列。例如，一个“产品需求分析 -> UI原型生成 -> 代码实现 -> 单元测试生成”的软件开发流水线。
2. 实现新的控制器：除了Claude Code，你可以实现OpenAIController、AnthropicController或甚至HumanController（将任务分配给真人）。只需继承BaseController并实现run方法，返回规范的ControllerEvent流。
3. 开发领域技能包：将你所在领域的专业知识（如法律文书分析、金融报告生成、生物信息学流程）封装成技能，形成一个可共享的技能库。
4. 增强审批门：默认的审批门是基于验证或人工提示。你可以实现更复杂的自动审批逻辑，例如，集成一个代码质量扫描工具来审批代码阶段，或集成一个图表美观度模型来审批图表阶段。

AgentWorld提供的是一套坚固的“乐高”积木。它的价值在于，让你能专注于构建智能体应用的业务逻辑和领域知识，而无需重复解决状态管理、可靠性、协作协议这些底层难题。从一个小型的概念验证开始，逐步叠加复杂度和可靠性，是使用这个平台的最佳路径。