企业官网建设流程全解析

LightAgent是一个轻量级 Python 智能体框架，v0.11 新增LightDAG运行时，支持在任务图未知的前提下让多个 Agent 围绕一张持久化任务图协作、并行处理独立节点、通过验证后发布成果，并可在进程重启后安全恢复执行。本文面向新手和普通用户，带你快速理解 LightDAG 的设计思路、核心概念与常见使用场景。

一、LightDAG 是什么？为什么需要它？

在传统多智能体系统中，任务流往往被预先定义好，一旦执行过程中发现需要拆分子任务，就难以优雅地动态调整。LightDAG正是为解决这一痛点而生。

它的核心定位：

• 动态任务图：Worker 在执行中可提出"拆解提案"（DecompositionProposal），把一个大任务拆成多个可并行的小任务。
• 严格验证：只有Verifier明确判定pass，候选产物才能被"发布"（published），否则保留为待审状态，不会污染下游。
• 可恢复执行：所有图状态、尝试、事件、租约都持久化在 SQLite 中，进程崩溃或重启后能安全接管并继续推进。

💡LightDAG 与 LightFlow 的区别：如果你的任务图是预先确定的，用更轻量的 LightFlow 即可；只有当你需要在运行时动态扩展任务图、并需要持久化恢复能力时，才选择 LightDAG。

相关源码入口：LightAgent/dag/

二、核心角色与术语速览

LightDAG 中几个必须理解的概念：

🚫重要原则：模型回复一段"我完成了"的文本不等于任务完成。Worker 必须显式提交TaskOutcome，且需通过 Verifier 校验。

相关类型定义：LightAgent/dag/models.py

三、最小可运行示例（无需真实 API）

LightAgent 提供了完全离线可跑通的示例 example/14.dynamic_dag.py，演示了"根任务 → 拆解为 analysis/summary → 并行执行 → 验证失败并修复 → 集成验收"的完整链路。

直接执行：

PYTHONPATH=. python example/14.dynamic_dag.py

示例中的核心步骤（伪代码级理解）：

1. 创建LightDAG实例，注入SqliteTaskGraphStore与LocalArtifactStore。
2. 注册一个local-worker，它在遇到root任务时返回TaskOutcome.decompose(...)，把root拆为analysis和summary。
3. summary第一次尝试返回一个"错误候选"，触发验证失败；第二次修复后通过。
4. 调用dag.run("offline-demo")，最终analysis、summary、root三个任务都变成verified。

完整离线示例：example/14.dynamic_dag.py

# 精简版核心结构 result = dag.run("offline-demo") states = {t.spec.task_id: t.state.status.value for t in dag.list_tasks(result.run_id)} # {'root': 'verified', 'analysis': 'verified', 'summary': 'verified'}

四、动态拆解：Worker 如何扩展任务图？

LightDAG 最强大的能力是"运行中扩展任务图"。当一个 Worker 执行到某任务时，可以返回TaskOutcome.decompose(proposal)提交一份 DecompositionProposal：

• parent_task_id：提出拆解的父任务
• new_tasks：要新增的TaskSpec列表
• rationale：拆解理由（人工审计用）
• composition_contract：组合要求，例如"必须同时完成 A 和 B"
• expected_graph_revision：期望的图版本，用于 CAS（Compare-And-Swap）冲突检测

接受拆解是一个原子的 SQLite 事务，会同时校验：

• 任务数/边数是否超过DAGConfig上限
• 是否存在未知依赖
• 是否存在环路
• 当前 attempt 租约是否仍有效
• 图版本是否与expected_graph_revision一致

只有全部通过，新节点才真正"存在"，父任务进入waiting_dependencies状态；等所有依赖都verified后，父任务以phase=integrate再次运行，进行集成验收。

⚠️不可变契约：TaskSpec的goal、acceptance_contract创建后不可原地改写。需要变更时，应通过 LightAgent/dag/scheduler.py 中的supersede_task()创建替代任务，并显式记录supersedes_task_id。

五、严格验证与产物发布

Verifier是 LightDAG 的"守门员"，只有它能决定一个任务是否verified。

5.1 验证报告必须绑定上下文

每份 VerificationReport 都会记录：

• 输入快照哈希（input_snapshot_hash）
• Verifier 的 key / version / config_digest
• 产物哈希列表
• 保障等级（AssuranceLevel：formal/executable_checks/human_review）
• 诊断信息与证据引用

只有verdict == pass的报告，才能把 staged 产物升级为published，任务标记为verified。

5.2 本地产物库

LocalArtifactStore 使用内容寻址（content-addressed）写入不可变 blob：

• 读取时强制检查tenant_id/project_id作用域
• 校验字节长度、SHA-256 完整性、路径包含性
• 未发布（staged）产物对下游不可见

📌生产建议：显式传入SqliteTaskGraphStore与LocalArtifactStore的持久化路径，避免默认落在.lightagent/artifacts。

dag = LightDAG( store=SqliteTaskGraphStore("data/dag.sqlite3"), artifact_store=LocalArtifactStore("data/artifacts"), verifiers={"exact": CallableVerifier("exact", verify_fn)}, config=DAGConfig(max_concurrency=2), )

六、可恢复执行：租约、幂等与接管

LightDAG 的"可恢复"不是简单保存 checkpoint，而是一整套租约 + 幂等 + 版本控制机制：

6.1 每次运行都持久化

Run 存储内容包含：任务图、任务状态、尝试历史、事件流、验证报告、产物清单、幂等记录、调度器租约。

6.2 租约（Lease）与 fencing

• 调度器持有运行级租约，默认 60 秒
• 每个任务尝试有独立 fencing epoch
• 旧 attempt 在租约过期或任务被替代后，无法再发布结果

6.3 恢复命令

result = dag.run(run_id) # 终端/暂停/阻塞状态的 run 不变 dag.resolve_blocker(run_id, task_id) # 显式人工解除阻塞 result = dag.run(run_id) dag.pause(run_id) result = await dag.resume(run_id) dag.cancel(run_id)

恢复时若发现DAGConfig不匹配，会以LA-DAG-RECOVERY-CONFIG-MISMATCH报错关闭（fail closed），避免用错误配置继续推进。

🛡️at-least-once 语义：fencing 只能保护 LightDAG 内部状态，无法撤销已发生的外部副作用。对外的关键操作（发请求、写数据库）应使用幂等设计或接入 Sandbox/Provider。

相关持久化实现：LightAgent/dag/store.py、LightAgent/dag/scheduler.py

七、安全边界：SecurityContext 与能力门

LightDAG 与 v0.10 的Unified Security Context深度融合，每个 run 都要求提供完整的 SecurityContext：

• user_id/tenant_id/project_id：三元组标识
• permissions：PermissionSet，声明允许的权限键（如dag.task.execute）
• Worker 子上下文只能收窄、不能扩展，包含 run、task、attempt、agent、资源、权限、网络策略、沙箱要求、截止时间、策略版本

CapabilityGate会在 Worker 创建前检查dag.task.execute能力，即使任务未声明资源也强制检查。

ApprovalToken则是操作级的：绑定操作、参数、身份、资源、策略版本，默认单次使用、可过期，不是通用 bearer token。

八、可观测性与 Tracing 集成

LightDAG 的每次事件都会写入 SQLite 事件流，可通过 Session/Trace 关联到run_id/task_id/attempt_id，配合 LightAgent 的 tracing 能力，可以在 Langfuse 等可观测平台看到完整的执行链路：

九、v0.11 的边界与适用场景

首版交付：

• 单机、单活动调度器、多异步 Worker
• 默认最多 4 个并发 Worker（DAGConfig.max_concurrency）
• SQLite 必须位于本地文件系统
• 完整的持久化、租约、幂等、恢复

首版不交付：

• 分布式队列（Redis/Celery）
• Web UI
• 自动 Git 合并/部署
• 通用代码沙箱
• 形式化证明工具

适合场景：

• 需要在运行时动态拆解任务的研究/数据流水线
• 对产物正确性要求严格、不能"模型说自己对就算对"的合规场景
• 长时任务需要进程重启后恢复，例如批处理报告、多阶段分析

不适合场景：

• 任务图固定不变的简单流水线 → 用 LightFlow
• 分布式大规模集群调度 → 需自行扩展 Provider

完整设计文档：docs/lightdag.md、docs/lightdag_v011_development_plan.zh-CN.md

十、快速上手清单 ✅

1. 安装：pip install LightAgent（或 clone 仓库本地安装）
2. 读文档：docs/lightdag.md 是最短路径
3. 跑示例：PYTHONPATH=. python example/14.dynamic_dag.py
4. 换你的 Worker：把示例中的local-worker替换成真实的 LightAgent 或业务函数
5. 加 Verifier：为每个verifier_key注册一个确定性或 LLM 校验函数
6. 显式持久化：生产环境指定SqliteTaskGraphStore和LocalArtifactStore路径
7. 绑定 SecurityContext：为每个 run 传入 tenant/project 身份与权限集
8. 观察日志：接入 Langfuse 等 tracing 平台，跟踪每次 attempt

总结

LightAgent v0.11 的LightDAG是"动态 + 持久化 + 严格验证"的三合一运行时。它把多智能体协作中"任务图不确定、产物可信度难保证、进程中断难恢复"三大痛点，通过版本化拆解提案、Verifier 守门、SQLite 租约与幂等三件套系统化解决。

如果你的工作流只需要固定步骤，LightFlow 更简单；如果你需要让 Agent在运行中自主规划并可靠完成，LightDAG 就是你要的下一步。

核心入口速查

• 运行时：LightAgent/dag/scheduler.py
• 数据契约：LightAgent/dag/models.py
• 持久化：LightAgent/dag/store.py
• 产物库：LightAgent/dag/artifacts.py
• 验证器：LightAgent/dag/verification.py
• Worker 适配：LightAgent/dag/worker.py
• 离线示例：example/14.dynamic_dag.py
• 官方文档：docs/lightdag.md