AI Agent 框架：工作流与 Agent 运行时模块设计

这里的关键约束是 Send + Sync，这意味着 Service 可以在多线程环境中安全共享和传递。Flow 包含了当前执行的上下文信息，Output 是标准化的返回结果。

执行计划

执行计划定义了 Service 的执行顺序和逻辑分支。我们需要按照计划不断执行 Service，并支持动态更新。

pub trait Plan: Send + Sync {
    /// 获取下一个节点
    fn next(&self, ctx: Arc<Context>, node_id: &str) -> NextNodeResult;
    
    /// 设置节点列表
    fn set(&self, nodes: Vec<PlanNode>);
    
    /// 动态更新节点配置
    fn update(
        &self,
        node_code: &str,
        update: Box<dyn FnOnce(Option<&mut PlanNode>) -> anyhow::Result<()>>,
    ) -> anyhow::Result<()>;
}

next 方法负责根据当前上下文决定下一步走向，支持条件分支和循环。update 方法允许在运行时修改计划，这对于 Smartflow 场景至关重要。

上下文 Context

Context 理解为执行一个任务的具体消息，包括任务编码、堆栈信息、状态等。它支持父子上下文的概念，即在一个 Service 内部可以执行子计划，父子上下文共享堆栈和部分状态。

pub struct Context {
    /// 父任务编码，用于追踪调用链
    pub parent_code: Option<String>,
    /// 任务流名称
    pub code: String,
    /// 状态：0:init 1:running 2:success 3:error
    pub status: AtomicU8,
    /// 堆栈信息，保护并发访问
    pub stack: Arc<Mutex<ContextStack>>,
    /// 执行计划
    pub plan: Arc<dyn Plan>,
    /// 全局扩展字段，用于存储临时数据
    pub extend: Mutex<HashMap<String, Box<dyn Any + Send + Sync + 'static>>>,
    /// 结束时回调
    pub over_callback: Option<Mutex<Vec<Box<dyn FnOnce(Arc<Context>) + Send + Sync + 'static>>>>,
    /// 关联的运行时实例
    pub(crate) runtime: Arc<Runtime>,
}

每个 Context 记录自己归属的运行时，确保多租户环境下的隔离性。AtomicU8 保证了状态变更的原子性，避免竞态条件。

调度

调度器按照 Plan 不断执行 Service。这里做了中间层设计，每个 Service 都是'洋葱'结构，一层层进，一层层出。所有 Service 都是异步执行的，如果 Plan 存在并行结构，两个分支会同时执行。

fn exec_next_node(ctx: Arc<Context>, node_code: &str) {
    let nodes = ctx.plan.next(ctx.clone(), node_code);
    
    for i in nodes {
        match ctx.runtime.nodes.get(i.node_type_id.as_str()) {
            Some(service) => {
                // 将 service 和 middle 封装成一个 flow 执行
                let flow = Flow::new(i, ctx.clone(), middle);
                tokio::spawn(async move {
                    let result = flow.call().await;
                    match result {
                        Ok(_) => {
                            // 执行成功则继续执行下一个
                            Runtime::exec_next_node(ctx, code.as_str());
                        }
                        Err(e) => {
                            // 错误处理与恢复
                            handle_error(ctx, e);
                        }
                    }
                });
            }
            None => {
                log::error!("Unknown node type: {}", i.node_type_id);
            }
        }
    }
}

Service Layer

为了方便 Service 的实现，我们将 Service 再包装一层，提供自动 JSON 解析功能。这样入参和出参可以直接绑定到 Struct，减少样板代码。

pub trait ServiceLayer: Sync + Send {
    type Config;
    type Output;

    async fn call(
        &self,
        code: String,
        ctx: Arc<Context>,
        cfg: Self::Config,
    ) -> anyhow::Result<Self::Output>;
}

impl<T, In, Out> Service for LayerJson<T, In, Out>
where
    T: ServiceLayer<Config = In, Output = Out>,
    In: for<'a> Deserialize<'a> + Send + Sync,
    Out: Serialize + Send + Sync,
{
    async fn call(&self, flow: Flow) -> anyhow::Result<Output> {
        // 解析入参
        let cfg = serde_json::from_str::<In>(flow.config.as_str())?;
        // 调用执行过程
        let output = self.handle.call(flow.id, flow.ctx, cfg).await?;
        // 解析出参
        let raw = serde_json::to_value(&output)?;
        Ok(Output::new(raw))
    }
}

并发安全与性能优化

在 Rust 实现的 Runtime 中，并发安全是核心考量。通过使用 Arc 和 Mutex，我们确保了上下文在多线程间的共享访问安全。

1. 内存管理：使用 Arc 包裹 Context 和 Plan，允许多个协程持有引用而不增加拷贝开销。
2. 锁粒度：ContextStack 使用 Mutex 保护，但在高并发场景下，可考虑细粒度锁或无锁数据结构来减少竞争。
3. 错误传播：通过 anyhow::Result 统一错误类型，便于在异步任务间传递错误信息并进行统一处理。

扩展性与维护性

为了支持未来的业务扩展，Runtime 设计了以下机制：

• 插件化架构：Service 接口允许第三方注册新的节点类型。
• 热更新：Plan 的 update 方法支持在不重启服务的情况下调整流程逻辑。
• 监控集成：Context 中的堆栈信息可用于链路追踪，方便排查生产环境问题。

总结

本设计通过模块化拆分，实现了灵活且强大的 AI Agent 运行时。支持从简单线性流程到复杂多智能体协作的各种场景。基于 Rust 的高性能特性，该方案在保证并发安全的同时，提供了良好的扩展性和可维护性，为构建企业级 AI 应用奠定了坚实基础。