从零写 Agent 框架：Workflow 与 Runtime 模块设计

另一种方法是做双循环，还是这个 agent，一个循环是外界有输入后再调用 agent，另一个循环是定时唤起 agent 的主观意识。从而节能减排。

不管是哪种方式，我更倾向于从 agent 外部解决。而非 agent 本身。

总结

现实中应用肯定不局限于这几种情况，但通过一定的流程编辑基本都可以解决，只是复杂性会比较高。

Runtime 设计

一个 agent 的 runtime 大概可以看成是由这几部分构成：调度 + 节点服务 + 执行计划 + 上下文。

节点 Service

一个 service 可以理解成一个原子能力，比如 LLM，比如 tool，也可能是 function。

我们这里做一个抽象：

pub trait Service: Send + Sync {
    async fn call(&self, flow: Flow) -> anyhow::Result<Output>;
}

执行计划

这个也很好理解，我们需要按照这个计划不断执行 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<()>;
}

上下文 Context

这个所谓的上下文 Context 可以理解为 执行一个任务的具体消息，比如任务编码，堆栈信息，状态等。

• 这里有一个子上下文的概念，也就是说在一个 service 里面可以执行一个子计划，也就是能力无限的实现。并且父子上下文会共享堆栈。todo: 也会共享状态。
• 这里没有做共享设计，每个 Context 会记录自己归属的运行时。也就是每个运行时都是独立的，为了多租户设计的。

pub struct Context {
    pub parent_code: Option<String>,
    // 任务流名称
    pub code: String,
    // 状态
    pub status: AtomicU8, // 0:init 1:running, 2:success, 3:error
    // 堆栈信息
    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>,
}

调度

调度也很简单，就是按照 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()) {
            ... // 获取服务实例
        };
        // 将 service 和 middle 封装成一个 flow 执行
        let flow = Flow::new(i, ctx.clone(), middle);
        tokio::spawn(async move {
            let ctx = flow.ctx.clone();
            let result = tokio::spawn(async move {
                // 处理一些堆栈信息
                // 执行 flow
                if let Err(e) = flow.call().await {
                    // 错误处理
                } else {
                    // 执行成功则继续执行下一个
                    Runtime::exec_next_node(ctx, code.as_str());
                }
            })
            .await;
            // 检查错误，和故障恢复
        });
    }
}

Service Layer

为了方便 service 的实现，我们将 service 再包装一层。如下：

pub trait ServiceLayer: Sync + Send {
    // service 的配置类型
    type Config;
    // 输出结果
    type Output;
    // 执行过程
    async fn call(
        &self,
        code: String,
        ctx: Arc<Context>,
        cfg: Self::Config,
    ) -> anyhow::Result<Self::Output>;
}

再为这个包装层加一个自动 json 解析的实现。这样入参和出参可以直接绑定到 struct。

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 = match serde_json::from_str::<In>(node_config.as_str()) {
            Ok(c) => c,
            Err(e) => return Err(anyhow!("config parse error: {}", e)),
        };
        // 调用执行过程
        let output = self.handle.call(code, ctx, cfg).await?;
        // 解析出参
        let raw = match serde_json::to_value(&output) {
            Ok(o) => o,
            Err(e) => return anyhow::anyhow!("code[{}],output json error:{}", node_info, e).into(),
        };
        Output::new(raw).raw_to_ctx().ok()
    }
}

异常处理与恢复

在复杂的 Agent 运行环境中，异常处理至关重要。除了基本的 Result 传播外，还需要考虑以下策略：

1. 重试机制：对于网络波动或临时性 LLM 超时，应配置指数退避重试。
2. 熔断机制：当某个 Service 连续失败超过阈值，暂时停止该节点的调度，防止雪崩效应。
3. 状态回滚：在执行关键业务逻辑前保存 Checkpoint，若后续步骤失败，可尝试回滚至上一安全状态。

pub enum RetryStrategy {
    Fixed(u32),
    ExponentialBackoff { base_ms: u64, max_times: u32 },
}

impl RetryStrategy {
    pub async fn execute<F, Fut, T>(&self, func: F) -> anyhow::Result<T>
    where
        F: Fn() -> Fut,
        Fut: Future<Output = anyhow::Result<T>>,
    {
        // 实现重试逻辑
        todo!()
    }
}

性能考量

为了保证高并发下的稳定性，Runtime 设计还需关注以下性能点：

• 内存管理：使用 Arc 共享不可变数据，减少拷贝。注意 Mutex 的持有时间，避免长时间阻塞。
• 异步调度：利用 tokio 的多线程特性，将 IO 密集型任务（如 LLM 调用）与 CPU 密集型任务（如 Prompt 组装）分离。
• 资源隔离：不同租户或任务的 Context 应严格隔离，防止数据泄露或资源争抢。

测试

单元测试是保证 Runtime 稳定性的关键。我们需要验证 Plan 的执行顺序、Context 的状态流转以及 Service 的调用逻辑。

Mock Service 实现

首先定义一个用于测试的 Mock Service：

struct MockService;

#[async_trait]
impl Service for MockService {
    async fn call(&self, flow: Flow) -> anyhow::Result<Output> {
        println!("MockService executed with input: {:?}", flow.input);
        Ok(Output::new("mock_result"))
    }
}

执行流程测试

使用 tokio::test 编写集成测试，模拟一个简单的线性流程：

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_simple_flow() {
        let runtime = Arc::new(Runtime::default());
        let mut context = Context::new("test_flow_001", runtime.clone());
        
        // 注册 Mock Service
        runtime.register_service("mock", Arc::new(MockService));
        
        // 设置执行计划
        let plan = SimplePlan::new(vec!["mock".to_string()]);
        context.set_plan(plan);
        
        // 启动执行
        let result = Runtime::exec_next_node(context, "mock").await;
        
        assert_eq!(context.status.load(Ordering::SeqCst), 2); // success
    }
}

并发场景测试

验证并行分支的正确执行：

#[tokio::test]
async fn test_parallel_branches() {
    let runtime = Arc::new(Runtime::default());
    let context = Arc::new(Context::new("parallel_test", runtime.clone()));
    
    // 模拟两个并行节点
    let nodes = vec!["service_a", "service_b"];
    
    // 实际测试中需确保两个任务同时完成且互不干扰
    // ... 省略具体实现细节
}

结语

整体 runtime 模块的设计核心在于解耦与扩展性。通过将 Service、Plan 和 Context 抽象化，我们能够灵活应对从简单工作流到复杂多智能体协作的各种场景。未来的优化方向包括引入更细粒度的权限控制、支持动态热更新 Plan 以及增强对分布式环境的适配能力。

希望这篇关于 Agent 框架 Runtime 设计的文章能为你构建自己的 AI 系统提供有价值的参考。