agentflow-in-the-flow-optimization

name	agentflow-in-the-flow-optimization
title	In-the-Flow Agentic System Optimization for Effective Planning and Tool Use
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2510.05592
keywords	["agent planning","multi-turn RL","credit assignment","tool use","modular agents"]
description	Decompose agent work across four specialized modules (planner, executor, verifier, generator) coordinated via evolving memory. Use Flow-GRPO to convert multi-turn sparse-reward optimization into sequential single-turn updates with outcome broadcasting, achieving 4-15% accuracy gains on benchmarks while scaling better than monolithic agent policies.

In-the-Flow Agentic System Optimization (AgentFlow)

Core Concept

Monolithic agent policies struggle with long-horizon tool-use tasks and generalization to new scenarios. AgentFlow decomposes agent responsibilities across four specialized modules—planner, executor, verifier, generator—coordinated through an evolving memory. Flow-GRPO handles sparse multi-turn rewards by broadcasting outcomes to align local decisions with global success, enabling efficient credit assignment without reward decomposition.

Architecture Overview

• Modular Components: Planner (high-level strategy), Executor (action selection), Verifier (quality checks), Generator (text output)
• Evolving Memory: Persistent state tracks task context, tool outputs, and decision rationale across turns
• Flow-GRPO Algorithm: Converts multi-turn sparse-reward RL into sequential single-turn training by broadcasting episode outcomes
• Group-Normalized Advantages: Stabilizes training across heterogeneous tasks and reward signals
• Scalability: Demonstrates improved gains with model size (7B baseline, compared against GPT-4o)

Implementation Steps

1. Modular Agent Architecture

Design four specialized modules with clear responsibilities.

class ModularAgent:
    def __init__(self, model_backbone, memory_size=2048):
        # Shared backbone (7B model)
        self.backbone = model_backbone

        # Specialized modules
        self.planner = PlannerModule(self.backbone)        # Strategy generation
        self.executor = ExecutorModule(self.backbone)      # Tool selection
        self.verifier = VerifierModule(self.backbone)      # Verification
        self.generator = GeneratorModule(self.backbone)    # Text generation

        
        .memory = Memory(max_size=memory_size)

     ():
        

        
        plan = .planner.generate_plan(
            task_instruction,
            current_observation,
            memory=.memory
        )

        
        action = .executor.select_action(
            plan,
            available_tools,
            memory=.memory
        )

        
        observation = execute_action(action)

        
        is_correct = .verifier.verify(
            action,
            observation,
            plan,
            memory=.memory
        )

        
        response = .generator.generate_text(
            plan, action, observation, is_correct,
            memory=.memory
        )

        
        .memory.append({
            : plan,
            : action,
            : observation,
            : is_correct,
            : response
        })

         response, observation

 :
     ():
        .buffer = []
        .max_size = max_size

     ():
        
        .buffer.append(decision_record)
         (.buffer) > .max_size:
            .buffer.pop()

     ():
        
         (.buffer) > :
            
            old_records = .buffer[:-]
            summary = .compact_summary(old_records)
            .buffer = [summary] + .buffer[-:]
         .buffer