1 Introduction↩︎

Large Language Models (LLMs) have demonstrated significant performance across a wide range of tasks, including code generation, question answering, and mathematical reasoning [1]–[9]. However, because these models generate instance-specific solutions, they lack the structural consistency and transparency needed for reliable deployment, while also being difficult to adapt to similar tasks. Workflows that encode recurring algorithmic patterns provide a principled alternative, decomposing complex challenges into structured, manageable steps. However, manually designing such workflows requires significant expertise and effort, limiting their broader application.

To address this challenge, recent effort have focused on the automatic workflow generation [10]–[13]. Endowing LLMs with this strategy planning capability means lowering the barrier for complex task automation from requiring manual programming by experts to merely providing high-level task descriptions, thereby greatly liberating productivity. Nevertheless, representing the workflow as static graph [14] or neural network [15] in many of these methods limits the flexibility of generatable workflows.

A promising direction emerges from works like ADAS [16], AFlow [17], ScoreFlow [18] and FlowReasoner [19], which represent workflows as code (a structured combination of predefined operators), making the automatic generation of workflows more flexible and expressive, where operator is an encapsulation of common agentic operations introduced by  [17]. Within this code-based framework, current approaches adopt two different paradigms for workflow generation.

The first paradigm comprises task-level approaches, exemplified by ADAS [16] and AFlow [17], which formulate workflow generation as a search problem within predefined task-operator set combination. Both methods employ iterative search strategies, with ADAS using evolutionary algorithms and AFlow using Monte Carlo Tree Search (MCTS), to discover high-performing workflows through repeated refinement. However, this search-based paradigm inherently constrains them to specific task and predetermined operator set. When encountering new tasks or operators, these methods require complete re-optimization from scratch, incurring substantial computational costs [18].

Conversely, the second paradigm consists of instance-level approaches, exemplified by ScoreFlow [18] and FlowReasoner [19], which generate workflows tailored to individual problem instances in the task. Both methods dynamically construct workflows at inference time, with ScoreFlow leveraging gradient-based optimization to refine agentic workflows, and FlowReasoner employing reasoning chains distilled from advanced models to design query-specific multi-agent systems. While these instance-level methods excel at tailoring workflows to specific problem instances, this granularity comes at the cost of reusability and deployment efficiency. They cannot capture task-level patterns that recur across similar problem instances, leading to redundant workflow generation for each query. In deployment scenarios, this approach foregoes the benefits of having optimized, reusable workflow templates that could be consistently applied to entire task.

The limitations of existing paradigms underscore two fundamental challenges that must be addressed to achieve truly general-purpose automatic workflow generation. (1) How can we overcome the re-optimization requirement of task-level approaches when facing new domains (task-operator set combinations)? (2) How can we learn generalizable patterns that avoid redundant instance-level generation while adapting effectively to untrained domains?

To systematically overcome the challenges, we propose MetaFlow, which formulates workflow generation as a meta-learning problem. As illustrated in Figure 1, unlike task-level search-based methods that require expensive re-optimization for new task-operator set combinations, and instance-level methods that generate workflows for individual queries, MetaFlow learns to directly synthesize workflows from task descriptions and operator set specifications, enabling zero-shot generation through a single model inference.

To achieve robust zero-shot generalization, MetaFlow employs a two-stage training paradigm with diverse task-operator pairs. Adopting the code-based workflow representation from prior works[16]–[19], we first synthesize thousands of workflows using Qwen-Max [20] across four tasks and a single operator set to finetune Qwen3-8B [21], establishing the foundation for understanding how tasks and operators relate to workflow structures. Subsequently, we apply online reinforcement learning with GRPO [22] across expanded domains, where execution feedback on problem instances directly optimizes the generation policy. This training ensures the model learns generalizable workflow construction principles rather than memorizing patterns. At inference, MetaFlow zero-shot generates effective workflows for novel configurations with only a single forward pass.

Our main contributions are:

• Meta-learning Framework: We introduce MetaFlow, a novel approach that reformulates workflow generation from discrete search within fixed configurations to continuous learning across diverse task-operator set combinations. By conditioning workflow generation on task descriptions and operator specifications, our framework achieves strong zero-shot generalization to unseen domains without any re-optimization, reducing computational cost from thousands of API calls to a single model inference.

• Scalable Training Pipeline: We design a two-stage training framework combining supervised learning with online reinforcement learning, utilizing diverse domains to ensure robust generalization to untrained domains.

• Comprehensive Evaluation: Extensive experiments demonstrate that MetaFlow achieves competitive performance on in-domain benchmarks while exhibiting remarkable zero-shot generalization to out-of-domain task classes and operator sets, including solving programming problems with novel operator combinations never seen during training and solving question answering problem using the vector database search operator.

2 Related Works↩︎

2.1 Agentic Workflow↩︎

Agentic workflows decompose complex tasks into structured steps through predefined operators and dependencies [17]–[19]. Unlike autonomous agents that learn through environment interaction [14], [23], workflows provide interpretable and consistent execution patterns. Recent works adopt code-based representations for superior expressiveness [16]–[19], supporting applications in code generation, question answering, and mathematical reasoning [1]–[9]. However, manual workflow design remains a significant bottleneck requiring deep expertise.

2.2 Automatic Workflow Generation↩︎

Recent advances have explored automating workflow generation for improving LLM performance [11], [12], [17], [18], [24]. While some methods optimize prompts within fixed workflows [10], [25], we focus on optimizing workflow structures directly.

Current structural optimization follows two paradigms. Task-level approaches like ADAS [16] and AFlow [17] search for optimal workflows through evolutionary algorithms or MCTS, but require complete re-optimization for new domains. Instance-level methods including ScoreFlow [18] and FlowReasoner [19] generate query-specific workflows but fail to extract reusable patterns.

Our Methods, MetaFlow reformulates workflow generation as meta-learning over diverse task-operator combinations during training. Through two-stage optimization combining supervised learning with reinforcement learning, it achieves true zero-shot generation—producing effective workflows for novel domains via single model inference, eliminating both re-optimization and adaptation overhead.

3 Problem Definition↩︎

Existing works often formulate automatic workflow generation as optimization problems [26], [27], requiring separate optimizations for each task. We elevate this perspective by reformulating it as a meta-learning problem [28], [29]. To ground this formulation, we first define our core concepts:

• Problem Instance \(\mathsf{p}\): A single, concrete problem to be solved.

• Task \(\mathsf{C}\): A family of problem instances sharing a common structure and solution strategy (e.g., GSM8K mathematical reasoning, DROP reading comprehension).

• Operator Set \(\mathsf{Ops}\): A collection of fundamental, reusable operations [17] (e.g., Generate, Revise, Ensemble).

• Domain \(\mathsf{(C, Ops)}\): The combination of a task and an operator set, defining a complete problem-solving context.

Within this meta-learning framework, an LLM serves as the meta-learner (the planner \(\pi_\theta\)). Its core responsibility is to learn a meta-strategy that enables fast adaptation: given any domain \(\mathsf{(C, Ops)}\), the planner rapidly generates an efficient and reusable Workflow \(\mathsf{W}\)—a structured sequence of operators from \(\mathsf{Ops}\). When executed on any Problem Instance \(\mathsf{p}\) from task \(\mathsf{C}\), this workflow produces a high-quality Solution \(\mathsf{s}\).

Unlike traditional meta-learning approaches such as Model-Agnostic Meta-Learning (MAML) [28], which rely on gradient updates for adaptation, our method achieves fast adaptation through the synthesis of workflows without requiring any gradient-based fine-tuning.

As a meta-learning task, our goal is to optimize the meta-parameters \(\theta\) of the planner \(\pi_\theta\) such that the generated workflows maximize expected rewards across the domain distribution \(\mathcal{D}_{\mathsf{(C, Ops)}}\). We adopt the classic bi-level optimization framework [29]: \[\begin{align} J(\theta; \mathsf{C}, \mathsf{Ops}) &= \mathbb{E}_{\mathsf{W} \sim \pi_\theta(\cdot \mid \mathsf{C}, \mathsf{Ops})}[\mathbb{E}_{\mathsf{p} \sim \mathsf{C}}[R(\text{Exec}(\mathsf{W},\mathsf{p}))]] \\ \theta^* &= \arg\max_\theta \mathbb{E}_{\mathsf{(C, Ops)} \sim \mathcal{D}_{\mathsf{(C, Ops)}}}[J(\theta; \mathsf{C}, \mathsf{Ops})] \end{align}\] where:

• \(\mathsf{(C, Ops)} \sim \mathcal{D}_{\mathsf{(C, Ops)}}\): Sample a domain (task-operator pair) from the distribution.

• \(\mathsf{W} \sim \pi_\theta(\cdot \mid \mathsf{C}, \mathsf{Ops})\): The planner performs fast adaptation—given task description \(\mathsf{C}\) and operator specifications \(\mathsf{Ops}\), it generates a customized workflow \(\mathsf{W}\) without gradient updates.

• \(\mathsf{p} \sim \mathsf{C}\): Sample problem instances from task \(\mathsf{C}\) to evaluate the workflow.

• \(\text{Exec}(\mathsf{W},\mathsf{p})\): Execute workflow \(\mathsf{W}\) on problem \(\mathsf{p}\) to produce a solution.

• \(R(\cdot)\): Reward function evaluating solution quality (e.g., test pass rate, answer correctness).

The outer loop optimizes performance across the entire domain distribution \(\mathcal{D}_{\mathsf{(C, Ops)}}\) rather than on individual tasks or instances, driving the meta-learner \(\pi_\theta\) to acquire cross-domain generalization capabilities. This contrasts sharply with prior works:

• Instance-level methods (e.g., ScoreFlow [18], ComfyUI-R1 [26]) optimize workflows for individual problem instances: \(\arg\max_{\mathsf{W}_\mathsf{p}} \mathbb{E}_{\mathsf{p} \sim \mathsf{C}}[R(\text{Exec}(\mathsf{W}_\mathsf{p},\mathsf{p}))]\).

• Task-level methods (e.g., AFlow [17]) optimize a single workflow per task but require separate optimization for each new task-operator combination.

Our framework elevates to the meta-level, learning a planner that generalizes across tasks and operators: \(\mathsf{W} \sim \pi_\theta(\cdot \mid \mathsf{C}, \mathsf{Ops})\).

4 Methodology↩︎

Our MetaFlow framework aims to train a large language model through a two-stage learning process to automatically generate efficient and reusable workflows for given domains \(\mathsf{(C, Ops)}\), where \(\mathsf{C}\) represents a task and \(\mathsf{Ops}\) denotes the available operator set. The framework addresses a key challenge in automated workflow generation: how to enable a model to quickly adapt to new domains (task-operator combinations) without requiring extensive training data for each. This section presents our system architecture (Section 4.1) and the two-stage training algorithm combining supervised fine-tuning and reinforcement learning with verifiable rewards (Section 4.2).

4.1 MetaFlow Architecture↩︎

The core architecture of MetaFlow consists of two components: (1) a planner LLM \(\pi_\theta\) that generates workflows conditioned on domain specifications \(\mathsf{(C, Ops)}\), and (2) an execution-evaluation environment that orchestrates workflow execution using the MetaGPT framework [30], invokes operators on sampled problem instances, and computes verifiable reward scores based on correctness evaluation.

Workflow Representation: A workflow is represented as a structured script based on the MetaGPT framework [30], composed of a series of predefined operator calls. As illustrated in the left part of Figure 2, we adopt the foundational text-processing operators from AFlow [17], including Generate, Summarize, Revise, Ensemble, and Programmer for basic tool invocation. Building upon the same MetaGPT interface, we further introduce additional operators for advanced text processing (e.g., Decompose [lst:decompose_operator], SelfConsistency) and domain-specific tools (e.g., VectorSearch [lst:vector_search_operator]). This standardized operator interface serves two purposes: (1) supervised fine-tuning provides a cold start that ensures the model generates syntactically valid operator calls with higher probability, and (2) the introduction of new operators demonstrates the necessity of our meta-learning approach for efficient and transferable workflow generation across diverse operator sets. Implementation details of the operators are provided in Appendix 8.1 for reference.

Input-Output: As illustrated in the right part of Figure 2, the model input consists of two parts: Task Description [lst:gsm8k_task_type_description], which elucidates the characteristics of the target task \(\mathsf{C}\) and the input-output formats of problem instances; Operator Descriptions [lst:operator_description], which define the functions, parameters, and input-output formats of each available operator in \(\mathsf{Ops}\) (whether pre-defined or user-defined). The output [lst:output_template] is a structured Workflow aimed at efficiently solving all problem instances from task \(\mathsf{C}\) using operator set \(\mathsf{Ops}\). This design paradigm allows users to introduce new operators and tasks through natural language descriptions at inference time, enabling the model to rapidly adapt without additional training. Implementation details are provided in Appendix 8.2.

Execution & Evaluation Environment: To achieve end-to-end optimization, we build an automated environment. This environment receives a candidate workflow \(\mathsf{W}_i\) and a set of \(N\) problem instances \(\mathsf{p}_1,\dots,\mathsf{p}_N\) sampled from task \(\mathsf{C}\) as input. In the execution phase, the environment uses the MetaGPT framework to orchestrate the workflow [30]. Each operator in the workflow (such as Generate) completes its specific subtask by calling an external lightweight language model API (e.g., Qwen-Turbo or GPT-4o-mini-0718) and processes each problem instance. Upon completion, an automated evaluator module verifies the correctness of each execution result, for example, by running unit tests or comparing outputs with ground truth answers. Based on the evaluation results, the environment computes a verifiable reward score \(R(\mathsf{W}_i)\) for workflow \(\mathsf{W}_i\), calculated as the average success rate over \(N\) instances: \[\begin{align} R(\mathsf{W}_i) = \frac{1}{N} \sum_{j=1}^{N} \mathbb{I}(\text{is correct}(\text{Exec}(\mathsf{W}_i, \mathsf{p}_j)))\label{eq:reward} \end{align}\tag{1}\] where \(\mathbb{I}(\cdot)\) is the indicator function. This reward score \(R(\mathsf{W}_i)\) is then passed to the training algorithm as the basis for policy updates.

4.2 Training Algorithm↩︎

The training process consists of two stages: supervised fine-tuning (SFT) and Reinforcement Learning with Verifiable Reward (RLVR). This design leverages supervised data to establish syntactic correctness, then employs reinforcement learning to optimize workflow effectiveness, separating structural learning from performance optimization.

4.2.1 Phase One: SFT Initialization↩︎

To address the cold start problem in reinforcement learning, we first perform supervised fine-tuning (SFT) on the base LLM using a dataset of (\([\mathsf{C},\mathsf{Ops}], \mathsf{W}\)) pairs, which is the pairs of domain \([\mathsf{C},\mathsf{Ops}]\) and the corresponding workflow \(\mathsf{W}\). This phase teaches the model to generate syntactically correct workflows following the required template structure (Listing [lst:output_template]) and proper operator invocation patterns (Listing [lst:operator_description]). By ensuring the model \(\pi_\theta\) can produce well-formed code, the subsequent RL phase can focus solely on optimizing workflow effectiveness rather than learning basic syntax, significantly narrowing the search space and accelerating convergence. Details of the SFT dataset construction, including the four-stage pipeline and LoRA configuration, are provided in Appendix 9.

4.2.2 Phase Two: RLVR Optimization↩︎

After SFT, we employ policy gradient algorithms to perform end-to-end self-improvement on the planner \(\pi_\theta\). The core of this phase is the Reinforcement Learning with Verifiable Reward (RLVR) loop, with the complete algorithm presented in Algorithm 3. The specific process consists of the following steps:

1. Policy Sampling: For a task \(\mathsf{C}\) sampled from the training set, the planner \(\pi_\theta\) generates a batch of \(k\) candidate workflows \(\mathsf{W}_1,\mathsf{W}_2,\dots,\mathsf{W}_k\).

2. Execution & Reward Calculation: Each candidate workflow \(\mathsf{W}_i\) is tested in the execution and evaluation environment described in Section 4.1, obtaining its corresponding reward score \(R(\mathsf{W}_i)\) executed on \(p_1^{(i)},\cdots,p_N^{(i)}\) (Equation 1 ) that reflects generalization capability: \[\begin{align} C \in \mathcal{D}_{\text{train}} &\overset{\pi_\theta}{\implies} \{W_i|1\leq i\leq k\}\\ &\overset{\text{each }W_i}{\implies}\{[W_i, R(W_i)]|1\leq i\leq k\} \end{align}\] 3. Policy Update: We use this batch of \([W_i, R(W_i)]\) pairs to update the parameters of the planner \(\pi_\theta\) using the Group Relative Policy Optimization (GRPO) algorithm [22]. The core idea of GRPO is to use the average performance within the group as a baseline to estimate advantages, thereby avoiding training an independent value network. \[\begin{align} \text{advantage} = \hat{A}(\mathsf{W}_i) = R(\mathsf{W}_i) - \mu_R,\; \mu_R = \sum_{j=1}^{k}R(\mathsf{W}_j)/k \end{align}\] This group-relative advantages makes the optimization signal derive from whether the workflow performance is better or worse than the current batch’s average level, rather than an absolute, potentially noisy value estimate.

4. Variance Reduction: Evaluation Based on Common Random Numbers: The effectiveness of policy gradients largely depends on the accuracy of the advantage function \(\hat{A}(\mathsf{W}_i)\) estimation. In our GRPO method, the advantage is computed relative to the batch average reward \(\mu_R\). If each workflow \(\mathsf{W}_i\) in the batch is evaluated on a set of independently and randomly sampled problem instances, the variance of the reward \(R(\mathsf{W}_i)\) will include not only policy randomness but also environmental randomness. This additional variance propagates to \(\mu_R\) and \(\hat{A}(\mathsf{W}_i)\), producing noisier gradients and reducing learning efficiency. To address this issue, we adopt a classic variance reduction technique: Common Random Numbers (CRN) [31]. In specific implementation, we ensure that all \(k\) candidate workflows \(\mathsf{W}_1, \dots, \mathsf{W}_k\) in the same training batch are evaluated on the exact same set of problem instances \(\{\mathsf{p}_1, \dots, \mathsf{p}_N\}\). By fixing the random variable of problem instances when comparing workflows, we eliminate noise arising from differences in problem sampling. This keeps the expected value of \(R(\mathsf{W}_i) - R(\mathsf{W}_j)\) unchanged but significantly reduces its variance. Ultimately, this ensures that our computed relative advantages more accurately reflect the intrinsic performance differences between workflows, leading to a more stable policy update direction and accelerating model convergence.

5 Experiments↩︎

This section evaluates MetaFlow through systematic experiments addressing three core questions: (1) Can MetaFlow achieve competitive performance with single-inference generation, eliminating the computational overhead of per-instance optimization? (2) Does the framework generalize to out-of-distribution tasks and novel operators unseen during training? (3) How do design choices—task-level versus instance-level generation, simple versus complex workflows—affect the performance-cost trade-off? We present training configuration, operator integration experiments, zero-shot generalization results, and main performance analysis.

5.1 Training Configuration↩︎

Stage 1: Supervised Fine-Tuning (SFT) Initialization. We first conduct supervised fine-tuning (SFT) on the base model Qwen3-8B to teach the model \(\pi_\theta^{\text{Base}}\) to generate syntactically correct workflows (Listing [lst:output_template]) and narrow the search space. We construct an expert dataset containing approximately \(1,300\) high-quality (\([\mathsf{C},\mathsf{Ops}], \mathsf{W}\)) pairs, covering the tasks \(\mathcal{D}_{\text{train}}^{\text{SFT}}=\{\text{GSM8K, DROP, MBPP, Humaneval}\}\) with the corresponding operator set for each task \(\mathsf{Ops}^{\text{SFT}}=(\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble})\). The dataset is synthesized using Qwen-Max API through a four-stage pipeline (Appendix 9). As shown in Figure 4 (left and middle), we train for one epoch with batch size \(16\) using LoRA (rank-\(16\)) [32] to prevent mode collapse on this limited dataset (Appendix 9.2).

Stage 2: Reinforcement Learning with Verifiable Rewards (RLVR) Optimization. Following SFT initialization, we employ RLVR for end-to-end optimization of the planner \(\pi_\theta^{\text{SFT}}\) with the same task set \(\mathcal{D}_{\text{train}}^{\text{RLVR}}=\mathcal{D}_{\text{train}}^{\text{SFT}}=\{\text{GSM8K, DROP, MBPP, Humaneval}\}\). Critically, to enhance generalization to diverse operator configurations, we train with four different operator sets \(\{\mathsf{Ops}^{\text{RLVR},i}\}_{i=1}^4\) that progressively introduce novel operators (Programmer, Decompose) beyond the base SFT set, randomly sampling domain combinations \((\mathsf{C}, \mathsf{Ops})\) at each iteration. The training employs Algorithm 3 for \(137\) steps. To control training costs, the execution of generated workflow \(\mathsf{W}_i\) on problem instance \(\mathsf{p}_j\) calls the Qwen-Turbo API. As shown in Figure 4 (right), the average reward rapidly increases from \(0.67\) to approximately \(0.88\) within the first \(35\) training steps, then stabilizes around \(0.85\)–\(0.90\), demonstrating the effectiveness of GRPO optimization with CRN. We select the checkpoint at step \(100\) for evaluation. The details are provided in Appendix 10.

5.2 Low-Cost Operator Integration: Continuous Library Expansion↩︎

A key advantage of MetaFlow over task-level search methods [16], [17] is its ability to continuously expand the operator library without costly re-optimization. To validate this capability, we demonstrate a practical workflow: abstracting a manually-tested reasoning pattern into a reusable operator. Specifically, we introduce the SelfConsistency operator (see Appendix [lst:self_consistency_operator] for complete implementation), which encapsulates the Self-Consistency pattern [33] (parallel generation with majority voting), directly derived from one of our experimental baselines (CoT SC in Table ¿tbl:tab:main95results?). Critically, incorporating this operator requires no model retraining, only defining its natural language interface following the format described in Listing [lst:operator_description].

To demonstrate immediate operator utilization, we generate workflows for GSM8K mathematical reasoning after integrating SelfConsistency. Listing [lst:self_consistency_workflow] shows an example generated workflow that successfully invokes the newly integrated operator with parallel sampling (n=5) and similarity-based voting, demonstrating that MetaFlow can immediately compose effective workflows with novel operators through single-inference generation without any retraining.

Implications. This validates MetaFlow’s low-cost continuous extensibility: (1) Minimal integration cost: introducing a new operator requires only defining its interface, without model retraining or iterative search (hours for MCTS/evolutionary methods vs. minutes for MetaFlow); (2) Rapid abstraction of proven patterns: practitioners can quickly encapsulate manually-tested workflow logic discovered through experimentation into reusable operators; (3) Immediate high-quality generation: generating effective workflows with the new operator costs merely a single inference; (4) Scalable library growth: each new operator becomes immediately available across all tasks without proportional computational overhead, enabling continuous knowledge accumulation. This paradigm transforms workflow optimization from isolated task-specific searches into a scalable knowledge base where proven patterns become reusable primitives.

5.3 Zero-Shot Generalization to Novel Operators↩︎

To validate true out-of-distribution (OOD) generalization, we evaluate MetaFlow on HotpotQA [3] multi-hop question answering with the VectorSearch operator (Listing [lst:vector_search_operator]) entirely unseen during training. This tests two critical OOD dimensions: (1) domain shift from math/code reasoning to retrieval-based QA, and (2) novel operator requiring hybrid document and sentence-level vector search. We generate 100 candidate workflows with \(\mathsf{Ops} = \{\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble}, \texttt{VectorSearch}\}\) evaluated on 100 HotpotQA instances (Validation), compared against a CoT+RAG baseline using the same API (Qwen-Turbo).

Figure 5 reveals the characteristics of zero-shot workflow generation. The left panel shows substantial performance variance: while 31% of workflows fail (score=0.0, likely due to syntax errors or incorrect operator usage), successful workflows demonstrate strong performance, with the best achieving \(0.74\) accuracy. The middle panel tracks the average score across generated workflows, which converges to \(0.29\), below the\(0.46\) baseline due to the high failure rate inherent in zero-shot generation. However, the right panel demonstrates that the best-of-\(k\) selection strategy efficiently identifies high-quality solutions: the best score monotonically increases, surpassing the baseline at \(k=76\) workflows and reaching \(0.74\) (60.9% improvement) with the test set (another \(100\) instances) with the same trend of performance. This validates that moderate sampling suffices to discover effective workflows despite zero-shot exploration risks. The highest-scoring workflow (Listing [lst:hotpotqa_best_workflow] in Appendix 13) implements sophisticated multi-hop reasoning through iterative retrieval-generation cycles—extracting entities, performing two-stage document search with connection analysis, and synthesizing information across retrieved contexts—demonstrating MetaFlow’s ability to compose complex operator sequences for unseen tools. The results confirm our meta-learning approach: train once on diverse tasks, then rapidly adapt to new operators and domains without re-optimization.

5.4 Main Results and Analysis↩︎

Experimental Setup. We evaluate MetaFlow on benchmarks spanning question answering (HotpotQA, DROP: 1,000 instances each), code generation (MBPP, HumanEval: full sets), and mathematical reasoning (GSM8K: 1,000 instances, MATH: Level-5 problems), following ScoreFlow’s configuration with 1:4 train-test splits. We compare against two baseline categories: (1) manually designed workflows—IO, CoT [34], CoT SC [33], MedPrompt [35], MultiPersona [36], Self-Refine [37]; (2) automatically optimized workflows—ADAS [16], AFlow [17], ScoreFlow [18]. The Planner Qwen3-8B uses base operators {Generate, Summarize, Revise, Ensemble} with dynamic prompt rewriting, plus novel operators {Decompose, Programmer} for OOD testing (natural language descriptions provided at inference). For each task, we generate 20 candidate workflows, validate on 50 instances, and select the best for testing. All methods use GPT-4o-mini-0718 as executor and judge. Workflow execution is orchestrated by MetaGPT.

Performance Analysis. Table ¿tbl:tab:main95results? in Appendix 7 shows MetaFlow achieves 78.8 average accuracy, competitive with ScoreFlow (82.5), AFlow (78.3), and ADAS (73.1) while requiring only single-inference generation versus their resource-intensive per-instance optimization. Notably, MetaFlow matches or exceeds manually designed workflows and surpasses many automated methods on individual tasks (e.g., GSM8K: 93.8). The framework demonstrates strong performance across diverse domains—mathematical reasoning (GSM8K, MATH), reading comprehension (DROP), and code generation (MBPP)—validating our meta-learning approach’s cross-task generalization. All baseline scores are from ScoreFlow [18]. And FlowReasoner [19] achieves \(82.19\) on MBPP with GPT-4o-mini-0718 as executor.

Understanding the Performance Gap: Paradigm Differences and Trade-offs. Our average performance (78.8) trails ScoreFlow (82.5) by 3.7 points, which reflects fundamental differences in problem formulation rather than algorithmic limitations. This gap stems from three interacting factors. First, risk amplification from task-level generation: As a cross-task generator, MetaFlow produces one workflow \(\mathsf{W}_C\) per task—any syntax error or malformed operator call yields zero accuracy across all test instances. In contrast, ScoreFlow’s instance-level approach generates \(\mathsf{W}_p\) per problem—a failed generation affects only one data point. This risk materializes in our VectorSearch experiments (Figure 5 left): 31% of generated workflows fail to execute, directly penalizing task-level metrics. Second, the complexity-performance trade-off: While simple workflows dominate on tasks solvable by direct reasoning, this advantage inverts for complex tool integration scenarios. On HotpotQA with the novel VectorSearch operator (Section 5.3), the simple CoT+RAG baseline achieves 0.46 accuracy. MetaFlow’s best workflow—requiring sophisticated orchestration of retrieval, decomposition, and reasoning operators—reaches 0.74 (+60.9% improvement), demonstrating that complex operator composition becomes essential when tool complexity increases. Critically, even with best-of-20 validation selection, our total cost remains orders of magnitude lower than iterative search methods: 20 generations yield a reusable task-level workflow versus thousands of MCTS evaluations (AFlow) or \(N\) per-instance generations (ScoreFlow). Figure 5 (right) shows best-of-\(k\) scaling stabilizes around \(k \approx 80\), where marginal sampling cost (\(<\$1\) with Qwen3-8B) far outweighs other automatically designed workflows baselines.

6 Conclusion↩︎

We introduce MetaFlow, a meta-learning framework that trains language models to generate task-level workflows through a two-stage paradigm combining supervised fine-tuning with reinforcement learning across diverse task-operator combinations. Across benchmarks in question answering, code generation, and mathematical reasoning, MetaFlow achieves competitive performance (78.8 average accuracy) with single-inference generation, while demonstrating strong zero-shot generalization to novel operators and domains. The observed 31% syntax error rate in zero-shot generation highlights a key limitation of single-inference approaches—future work could explore multi-turn reinforcement learning where the model iteratively refines workflows through interaction with execution feedback, potentially combining the efficiency of learned meta-strategies with the robustness of adaptive generation.

Acknowledgements↩︎

This work was supported by the National Natural Science Foundation of China (Grant No. 625B1012).

7 Main Results↩︎

\begin{table}[ht]

-0.1in \end{table}

8 Metaflow Architecture↩︎

8.1 Operators Design↩︎

For the basic operators defined in the left part of Figure 2, their implementations are identical to those in the AFlow [17] codebase, and thus we omit their detailed descriptions here. In the following, we introduce the newly defined operators that we have designed for this work.

Decompose Operator: Beyond the basic text-processing operators, we introduce the Decompose operator to handle complex problem-solving scenarios that require hierarchical decomposition. This operator breaks down intricate problems into structured subproblems with explicit dependency relationships, enabling workflows to tackle multi-step reasoning tasks systematically. The operator is invoked as: await self.decompose(instruction: str, context: str) and returns a list of subproblem dictionaries, each containing an ID, description, and dependencies. The following code block shows the implementation details of this operator.

class Decompose(Operator):
    """
    Core Operator: Decompose
    Breaks down complex problems into manageable subproblems.
    """
    async def __call__(self, instruction: str = "", context: str = "") -> List[Dict[str, str]]:
        prompt = f"""You are an expert at problem decomposition. Break down the complex problem into manageable subproblems.

**Instruction on decomposition strategy:**
{instruction}

**Problem/Context to Decompose:**
{context if context else "No context provided."}

**Original Problem:**
{self.problem_text}

Your response MUST be valid XML with 'think' and 'subproblems' fields.
- In "think", explain your decomposition strategy
- In "subproblems", provide a list where each item has:
  - id: unique identifier (e.g., "sub1", "sub2")
  - description: clear description of the subproblem
  - dependencies: comma-separated IDs of prerequisite subproblems (empty if none)

**EXAMPLE:**
<think>This problem requires finding area then volume.</think>
<subproblems>
[
  {{"id": "sub1", "description": "Calculate the radius", "dependencies": ""}},
  {{"id": "sub2", "description": "Calculate the area", "dependencies": "sub1"}},
  {{"id": "sub3", "description": "Calculate the volume", "dependencies": "sub2"}}
]
</subproblems>"""
        
        response = await self._fill_node(DecomposeOp, prompt, mode="xml_fill")
        response = DecomposeOp(**response) 
        return [sub.dict() for sub in response.subproblems]

SelfConsistency Operator: To demonstrate MetaFlow’s capability for low-cost operator integration and continuous library expansion (Section 5.2), we introduce the SelfConsistency operator—a practical abstraction of the Self-Consistency reasoning pattern [33]. This operator encapsulates the parallel generation with majority voting strategy, which was originally one of our experimental baselines (CoT SC in Table ¿tbl:tab:main95results?). Critically, incorporating this operator into MetaFlow requires no model retraining; the planner learns to effectively utilize it through its natural language interface description alone. The following code block presents the complete implementation:

class SelfConsistency(Operator):
    """Self-Consistency: multi-path sampling + majority voting for better reasoning."""
    
    def __init__(self, llm, problem_text: str = "", n_samples: int = 5,
                 similarity_threshold: float = 0.85, return_full_info: bool = False):
        super().__init__(llm, problem_text)
        self.n_samples = n_samples
        self.similarity_threshold = similarity_threshold
        self.return_full_info = return_full_info
    
    async def __call__(self, instruction: str = "", context: str = "", answer_type: str = "auto") -> str:
        silent = os.environ.get('SCOREFLOW_SILENT', 'false').lower() == 'true'
        if not silent:
            print(f"[SelfConsistency] samples={self.n_samples}, threshold={self.similarity_threshold}")
        
        paths = await self._parallel_sampling(instruction, context)
        if not silent:
            print(f"  Generated {len(paths)} paths")
        
        raw_answers = await self._extract_answers_batch(paths, answer_type)
        if not silent:
            print(f"  Extracted {len([a for a in raw_answers if a])} answers")
        
        normalized = [self._normalize_answer(a, answer_type) for a in raw_answers]
        clustered = self._cluster_similar_answers(normalized)
        votes = Counter(clustered)
        
        if not votes:
            if not silent:
                print("  No valid answers, fallback")
            return await self._fallback_generate(instruction, context)
        
        answer, count = votes.most_common(1)[0]
        conf = count / len(clustered) if clustered else 0
        if not silent:
            print(f"  Votes: {dict(votes)} | Answer: {answer} | Conf: {conf:.1        
        if self.return_full_info:
            return self._format_full_response(paths, raw_answers, votes, answer, conf)
        return answer
    
    async def _parallel_sampling(self, instruction: str, context: str) -> List[str]:
        prompt = f"""Solve step by step.
**Problem:** {self.problem_text}
**Instruction:** {instruction or "Solve carefully."}
**Context:** {context or "None"}
Show reasoning. End with "Final Answer:" or "The answer is:"
**Solution:**"""
        tasks = [self._sample_single_path(prompt) for _ in range(self.n_samples)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return [r for r in results if isinstance(r, str) and len(r.strip()) > 10]
    
    async def _sample_single_path(self, prompt: str) -> str:
        try:
            return (await self._fill_node(GenerateOp, prompt, mode="single_fill")).get("response", "")
        except:
            return ""
    
    async def _extract_answers_batch(self, paths: List[str], answer_type: str) -> List[str]:
        answers = []
        for p in paths:
            ans = self._rule_based_extraction(p, answer_type)
            answers.append(ans if ans else await self._llm_extract_answer(p, answer_type))
        return answers
    
    def _rule_based_extraction(self, text: str, answer_type: str) -> str:
        patterns = [
            r"(?:final answer|the answer is|answer:)\s*[:\s]*(.+?)(?:\n|$|\.(?:\s|$))",
            r"(?:therefore|thus|so|hence),?\s*(?:the answer is)?\s*[:\s]*(.+?)(?:\n|$|\.(?:\s|$))",
            r"\\boxed\{(.+?)\}", r"\*\*(.+?)\*\*\s*$",
        ]
        for p in patterns:
            m = re.search(p, text.lower(), re.IGNORECASE | re.MULTILINE)
            if m:
                ans = re.sub(r'^[:\s]+|[:\s\.]+$', '', m.group(1).strip())
                if ans and len(ans) < 200:
                    return ans
        if answer_type == "choice":
            m = re.search(r"(?:answer|option|choice)[:\s]*([A-Da-d])\b", text, re.IGNORECASE)
            if m: return m.group(1).upper()
        if answer_type == "numeric":
            nums = re.findall(r'-?\d+(?:\.\d+)?(?:/\d+)?', text)
            if nums: return nums[-1]
        return ""
    
    async def _llm_extract_answer(self, path: str, answer_type: str) -> str:
        hints = {"numeric": "number only", "choice": "letter only", "short": "brief", "boolean": "yes/no only"}
        prompt = f"Extract final answer. {hints.get(answer_type, '')} Return ONLY the answer.\n**Solution:** {path[-2000:]}\n**Answer:**"
        try:
            ans = (await self._fill_node(GenerateOp, prompt, mode="single_fill")).get("response", "").strip()
            return re.sub(r'^(the answer is|answer:|final answer:)\s*', '', ans, flags=re.IGNORECASE).strip()
        except:
            return ""
    
    def _normalize_answer(self, answer: str, answer_type: str) -> str:
        if not answer: return ""
        n = re.sub(r'^(the |a |an )', '', answer.strip().lower())
        n = re.sub(r'[.,!?;:]+$', '', n)
        if answer_type == "numeric" or re.match(r'^-?\d', n):
            n = re.sub(r'[$            if '/' in n:
                try: p = n.split('/'); n = str(float(p[0]) / float(p[1])) if len(p) == 2 else n
                except: pass
        elif answer_type == "choice":
            m = re.search(r'[A-Da-d]', n)
            if m: n = m.group(0).upper()
        elif answer_type == "boolean":
            n = 'yes' if n in ['yes','true','1','correct','right'] else ('no' if n in ['no','false','0','incorrect','wrong'] else n)
        return n
    
    def _cluster_similar_answers(self, answers: List[str]) -> List[str]:
        valid = [a for a in answers if a]
        if not valid or self.similarity_threshold >= 1.0: return valid
        clusters = []
        for ans in valid:
            matched = False
            for rep, members in clusters:
                if SequenceMatcher(None, ans, rep).ratio() >= self.similarity_threshold:
                    members.append(ans); matched = True; break
            if not matched: clusters.append((ans, [ans]))
        return [Counter(m).most_common(1)[0][0] for a in valid for r, m in clusters if a in m]
    
    async def _fallback_generate(self, instruction: str, context: str) -> str:
        prompt = f"Solve: {self.problem_text}\nInstruction: {instruction}\nContext: {context or 'None'}"
        return (await self._fill_node(GenerateOp, prompt, mode="single_fill")).get("response", "No answer")
    
    def _format_full_response(self, paths: List[str], raw: List[str], votes: Counter, answer: str, conf: float) -> str:
        lines = [f"## Self-Consistency Result", f"**Final Answer:** {answer}", f"**Confidence:** {conf:.1                 "### Votes"] + [f"- {a}: {c}" for a, c in votes.most_common()]
        return "\n".join(lines)
}}

VectorSearch Operator: To evaluate the generalization capability of MetaFlow to completely out-of-distribution (OOD) operators, we introduce the VectorSearch operator [lst:vector_search_operator], which is a complex tool-calling operator that was entirely unseen during both the SFT and RLVR training phases. Unlike the basic text-processing operators used in training, VectorSearch requires sophisticated external API interactions with a vector database (ChromaDB) and involves multiple parameters for controlling retrieval behavior. The operator is invoked as: await self.vector_search(instruction: str, context: str, top_k: int), where it performs hybrid retrieval combining document-level and sentence-level semantic search over the HotpotQA knowledge base. Despite its complexity and complete absence from training data, MetaFlow successfully learns to incorporate this operator into generated workflows based solely on its natural language description provided at inference time, as demonstrated in Appendix 13.

class VectorSearch(Operator):
    """
    Core Operator: Vector Search
    Retrieves relevant documents from HotpotQA vector database using RAG system.
    """

    def __init__(self, llm, problem_text: str = "", db_config: Dict = None):
        super().__init__(llm, problem_text)
        self.config = self._load_config()
        if db_config:
            self.config.update(db_config)
        self._init_chromadb()

    def _load_config(self) -> Dict:
        """Load configuration from db.config file"""
        from pathlib import Path
        config = {}
        config_file = Path(__file__).parent / "db.config"

        if config_file.exists():
            with open(config_file, 'r') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#') and '=' in line:
                        key, value = line.split('=', 1)
                        key, value = key.strip(), value.strip()
                        if key in ('DOC_TOP_K', 'SENT_TOP_K'):
                            config[key.lower()] = int(value)
                        elif key == 'HYBRID_MODE':
                            config[key.lower()] = value.lower() == 'true'
                        else:
                            config[key.lower()] = value
        else:
            config = {
                'db_path': '...chroma_db', 'model_path': '...all-MiniLM-L6-v2',
                'doc_top_k': 3, 'sent_top_k': 5, 'hybrid_mode': True
            }
        return config

    def _init_chromadb(self):
        """Initialize ChromaDB connection and collections"""
        try:
            import chromadb
            from chromadb.utils import embedding_functions
            self.client = chromadb.PersistentClient(path=self.config['db_path'])
            model = self.config['model_path'] if os.path.exists(self.config['model_path']) else "all-MiniLM-L6-v2"
            self.embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(model_name=model)
            self.doc_collection = self.client.get_collection("documents")
            self.sent_collection = self.client.get_collection("sentences")
        except Exception as e:
            self.doc_collection = self.sent_collection = None

    async def __call__(self, instruction: str = "", context: str = "", top_k: int = None) -> str:
        """Execute vector search and return formatted retrieved documents."""
        if not self.doc_collection or not self.sent_collection:
            return "Error: Vector database not initialized."
        doc_k = top_k or self.config['doc_top_k']
        query = await self._process_query(instruction, context)
        retrieved_data = self._hybrid_retrieval(query, doc_k, self.config['sent_top_k'])
        return self._format_context(retrieved_data)

    async def _process_query(self, instruction: str, context: str) -> str:
        """Process and combine query from instruction and context."""
        if instruction and context:
            return f"{instruction} {context}"
        return instruction or (context[:200] if context else self.problem_text[:200] or "general information")

    def _hybrid_retrieval(self, query: str, doc_k: int, sent_k: int) -> Dict:
        """Perform hybrid document and sentence level retrieval."""
        retrieved_data = {'documents': [], 'scores': []}
        try:
            if self.config.get('hybrid_mode', True):
                doc_results = self.doc_collection.query(query_texts=[query], n_results=doc_k)
                for i, (doc_id, doc_text, meta, dist) in enumerate(zip(
                    doc_results['ids'][0], doc_results['documents'][0], 
                    doc_results['metadatas'][0], doc_results['distances'][0])):
                    retrieved_data['documents'].append({
                        'doc_id': doc_id, 'title': meta.get('title', 'Unknown'),
                        'text': doc_text[:500], 'type': 'document', 'rank': i + 1})
                    retrieved_data['scores'].append(float(dist))
            sent_results = self.sent_collection.query(query_texts=[query], n_results=sent_k)
            for i, (sid, stxt, meta, dist) in enumerate(zip(
                sent_results['ids'][0], sent_results['documents'][0],
                sent_results['metadatas'][0], sent_results['distances'][0])):
                if i < 3:
                    retrieved_data['documents'].append({
                        'doc_id': sid, 'title': meta.get('title', 'Unknown'),
                        'text': stxt, 'type': 'sentence', 'rank': i + 1})
                    retrieved_data['scores'].append(float(dist))
        except Exception:
            pass
        return retrieved_data

    def _format_context(self, retrieved_data: Dict) -> str:
        """Format retrieved documents into readable context string."""
        if not retrieved_data['documents']:
            return "No relevant documents found."
        parts = ["**Retrieved Information:**\n"]
        for d in retrieved_data['documents']:
            parts.append(f"[{d['type'].title()}: {d['title']}] {d['text']}\n")
        return "\n".join(parts)

8.2 Input and Output of the Model↩︎

The model input consists of two parts: (1) Task Type Description, which elucidates the domain characteristics of the target problem family and the input-output formats of each belonging problem instance. The block below shows the complete Task Type Description for GSM8K.

### Problem Domain Overview
This domain tests multi-step mathematical reasoning using basic arithmetic operations.

#### Key Characteristics & Requirements
- **Answer Format**: Single numerical value (integer or decimal)
- **Solution Steps**: 2-8 step reasoning chains using +, -, *, /
- **Critical**: Track intermediate results and units throughout
- **Validation**: Final answer must be numerically exact
- **No Complex Math**: Only elementary arithmetic, no algebra or calculus

#### Common Problem Types & Solution Strategies
- **Sequential Operations**: Step-by-step calculations building on previous results
- **Rate Problems**: Distance/speed/time, work rates, unit prices
- **Distribution**: Dividing quantities, equal sharing, remainders
- **Proportions**: Percentages, fractions, ratios, scaling
- **Multi-entity**: Track different quantities for multiple people/objects

#### Workflow Focus Points
1. Extract all numerical values and their context
2. Identify what the question asks for
3. Build step-by-step calculation chain
4. Show intermediate results explicitly
5. Return final numerical answer only

#### Input Format
```
---
**QUESTION:**
[Complete word problem text]
---
```
Multiple problems follow the same structure if provided.

(2) Operator Descriptions, which define the functions, parameters, and input-output formats of each available operator (whether pre-set or user-defined). The block below shows the complete Operator Descriptions of the operator set \(\mathsf{Ops} =\) (Generate, Summarize, Revise, Ensemble).

### Available Operators & Building Blocks
All operators follow a consistent interface pattern and are initialized with the problem text.

#### **Core Operators**
**Generate: CREATE new information**
- **Signature:** `await self.generate(instruction: str, context: str = "") -> str`
- **Purpose:** Produces new text, analysis, or reasoning based on strategic instructions

**Revise: IMPROVE existing information**
- **Signature:** `await self.revise(instruction: str, context: str) -> str`
- **Purpose:** Critiques and refines existing text based on specific improvement criteria

**Summarize: COMPRESS information**
- **Signature:** `await self.summarize(instruction: str, context: str) -> str`
- **Purpose:** Condenses text while preserving key information relevant to the problem

**Ensemble: DECIDE between or synthesize options**
- **Signature:** `await self.ensemble(instruction: str, contexts: List[str]) -> str`
- **Purpose:** Evaluates, compares, or merges multiple candidate solutions

The output requires a structured Workflow aimed at efficiently solving all problem instances under the task type using the given operator set based on the template below.

```python
# --- DO NOT IMPORT HERE ---
class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)
        
        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        """
        import asyncio
        # --- YOUR WORKFLOW LOGIC HERE ---
```

9 Details of Supervised Fine-Tuning↩︎

The supervised fine-tuning (SFT) stage initializes the planner model \(\pi_\theta\) to generate syntactically correct workflows following the required template structure (see Appendix 8.2 for the detailed input-output format). This stage addresses the cold start problem by teaching the model the basic grammar of workflow construction before reinforcement learning optimization.

Dataset Construction Pipeline. We construct approximately 1,300 high-quality (\([\mathsf{C},\mathsf{Ops}], \mathsf{W}\)) pairs using Qwen-Max as the expert model. The construction follows a four-stage pipeline:

1. Prompt Crafting: Construct input prompts combining task descriptions \(\mathsf{C}\) (domain characteristics, input-output formats) and operator specifications \(\mathsf{Ops}\) (function signatures, purposes) following the format defined in Appendix 8.2.

2. Expert Generation: Feed prompts to Qwen-Max API to obtain comprehensive responses including workflow explanations and complete executable code.

3. Code Extraction: Parse API responses to extract clean workflow code \(\mathsf{W}\), discarding natural language explanations.

4. Quality Verification: Execute each extracted workflow on 1-2 randomly sampled problem instances \(\mathsf{p} \sim \mathsf{C}\) to ensure (a) no syntax errors and (b) correct solutions. Only validated workflows are retained.

This pipeline is applied across four tasks (GSM8K, DROP, MBPP, Humaneval) with the basic operator set \(\mathsf{Ops}^{\text{SFT}} = (\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble})\), producing diverse workflow patterns that establish the foundation for RLVR optimization. To ensure stable training on this limited dataset, we employ parameter-efficient fine-tuning with LoRA configuration, as detailed in Section 9.2.

9.1 Dataset Construction Examples↩︎

To illustrate the construction pipeline, we present a concrete example for the GSM8K mathematical reasoning task. The input prompt (Listing [lst:sft_input_prompt]) combines task description and operator specifications:

### 1. Problem Domain Overview
This domain tests multi-step mathematical reasoning using basic arithmetic operations.

#### Key Characteristics & Requirements
- **Answer Format**: Single numerical value (integer or decimal)
- **Solution Steps**: 2-8 step reasoning chains using +, -, *, /
- **Critical**: Track intermediate results and units throughout
- **Validation**: Final answer must be numerically exact
- **No Complex Math**: Only elementary arithmetic, no algebra or calculus

#### Common Problem Types & Solution Strategies
- **Sequential Operations**: Step-by-step calculations building on previous results
- **Rate Problems**: Distance/speed/time, work rates, unit prices
- **Distribution**: Dividing quantities, equal sharing, remainders
- **Proportions**: Percentages, fractions, ratios, scaling
- **Multi-entity**: Track different quantities for multiple people/objects

#### Workflow Focus Points
1. Extract all numerical values and their context
2. Identify what the question asks for
3. Build step-by-step calculation chain
4. Show intermediate results explicitly
5. Return final numerical answer only

### 2. Available Operators & Building Blocks
All operators follow a consistent interface pattern and are initialized with the problem text.

#### **Core Operators**
**Generate: CREATE new information**
- **Signature:** `await self.generate(instruction: str, context: str = "") -> str`
- **Purpose:** Produces new text, analysis, or reasoning based on strategic instructions

**Revise: IMPROVE existing information**
- **Signature:** `await self.revise(instruction: str, context: str) -> str`
- **Purpose:** Critiques and refines existing text based on specific improvement criteria

**Summarize: COMPRESS information**
- **Signature:** `await self.summarize(instruction: str, context: str) -> str`
- **Purpose:** Condenses text while preserving key information relevant to the problem

**Ensemble: DECIDE between or synthesize options**
- **Signature:** `await self.ensemble(instruction: str, contexts: List[str]) -> str`
- **Purpose:** Evaluates, compares, or merges multiple candidate solutions

#### Input Format
```
---
**QUESTION:**
[Complete word problem text]
---
```
Multiple problems follow the same structure if provided.

### 3. Your Task: Complete the `run_workflow` Method
Your task is to write the Python code for the `run_workflow` method within the provided template. 
Focus on creating a robust, reusable workflow that leverages detailed instructions.

**Base Template:**
```python
# --- DO NOT IMPORT HERE ---
class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)
        
        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        """
        import asyncio
        # --- YOUR WORKFLOW LOGIC HERE ---
```

Given this input prompt, Qwen-Max generates a comprehensive response that includes both natural language explanation of the workflow design and the complete workflow code. Listing [lst:sft_qwen_output] shows the raw output from the API:

Now let me understand the task better and implement the `run_workflow` method.
This is a workflow for solving multi-step mathematical reasoning problems using the provided operators.

I've created a robust workflow for multi-step mathematical reasoning problems. Here's how it works:

(*@\texttt{```python}@*)
# --- DO NOT IMPORT HERE ---
class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)
        
        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        """
        # Step 1: Extract all numerical values and their context
        extracted_info = await self.generate(
            instruction="Extract all numerical values from the problem, including their units and contextual meaning (e.g., '5 apples', 'speed of 60 km/h'). List them clearly.",
            context=self.problem_text
        )
        
        # Step 2: Identify what the question asks for
        target_query = await self.generate(
            instruction="Based on the problem, state precisely what numerical value is being asked for, including its expected unit if any.",
            context=self.problem_text
        )
        
        # Step 3: Build a step-by-step calculation plan
        plan = await self.generate(
            instruction="Create a clear, step-by-step plan (2 to 8 steps) to solve the problem using only +, -, *, / operations. Reference the extracted numbers and specify what each step computes.",
            context=f"Extracted info: {extracted_info}\nTarget: {target_query}"
        )
        
        # Step 4: Execute the plan with explicit intermediate results
        calculation = await self.generate(
            instruction="Perform the arithmetic step-by-step as per the plan. Show each intermediate result with its unit or meaning. Ensure only elementary arithmetic is used.",
            context=plan
        )
        
        # Step 5: Extract and validate the final numerical answer
        final_answer_raw = await self.generate(
            instruction="From the calculation steps, extract only the final numerical answer (as an integer or decimal). Do not include units, explanations, or text -- only the number.",
            context=calculation
        )
        
        # Step 6: Revise to ensure numerical exactness and proper format
        final_answer_clean = await self.revise(
            instruction="Ensure the output is a single numerical value (e.g., 42 or 15.75), with no extra characters, units, or formatting. If it's a whole number, do not include decimal places unless required.",
            context=final_answer_raw
        )
        
        # Step 7: Return the final answer
        return final_answer_clean.strip()
(*@\texttt{```}@*)

**Workflow Design:**

1. **Problem Extraction** - Uses `generate` to parse the problem, extracting all numerical values, the goal, and relationships between quantities

2. **Parallel Solution Generation** - Generates three independent solutions using different approaches:
   - Standard step-by-step forward calculation
   - Alternative verification method
   - Working backwards from the goal

3. **Solution Revision** - Each solution is revised to catch arithmetic errors, verifying each calculation step

4. **Ensemble Decision** - Compares all three verified solutions to determine the correct answer through majority vote or error analysis

5. **Answer Extraction** - Uses `summarize` to extract just the final numerical value

**Key Features:**
- Uses `asyncio.gather` for parallel execution (generating and revising solutions simultaneously)
- Multiple independent solutions reduce the chance of systematic errors
- Revision step catches arithmetic mistakes
- Ensemble aggregates results for higher accuracy
- Clean numerical output extraction at the end

Finally, we parse the API response to extract only the executable workflow code, discarding the natural language explanations. This extracted code forms a single training example in our SFT dataset. Listing [lst:sft_extracted_workflow] shows the clean workflow code extracted from the above output:

# --- DO NOT IMPORT HERE ---
class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)
        
        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        """
        # Step 1: Extract all numerical values and their context
        extracted_info = await self.generate(
            instruction="Extract all numerical values from the problem, including their units and contextual meaning (e.g., '5 apples', 'speed of 60 km/h'). List them clearly.",
            context=self.problem_text
        )
        
        # Step 2: Identify what the question asks for
        target_query = await self.generate(
            instruction="Based on the problem, state precisely what numerical value is being asked for, including its expected unit if any.",
            context=self.problem_text
        )
        
        # Step 3: Build a step-by-step calculation plan
        plan = await self.generate(
            instruction="Create a clear, step-by-step plan (2 to 8 steps) to solve the problem using only +, -, *, / operations. Reference the extracted numbers and specify what each step computes.",
            context=f"Extracted info: {extracted_info}\nTarget: {target_query}"
        )
        
        # Step 4: Execute the plan with explicit intermediate results
        calculation = await self.generate(
            instruction="Perform the arithmetic step-by-step as per the plan. Show each intermediate result with its unit or meaning. Ensure only elementary arithmetic is used.",
            context=plan
        )
        
        # Step 5: Extract and validate the final numerical answer
        final_answer_raw = await self.generate(
            instruction="From the calculation steps, extract only the final numerical answer (as an integer or decimal). Do not include units, explanations, or text -- only the number.",
            context=calculation
        )
        
        # Step 6: Revise to ensure numerical exactness and proper format
        final_answer_clean = await self.revise(
            instruction="Ensure the output is a single numerical value (e.g., 42 or 15.75), with no extra characters, units, or formatting. If it's a whole number, do not include decimal places unless required.",
            context=final_answer_raw
        )
        
        # Step 7: Return the final answer
        return final_answer_clean.strip()

The above example demonstrates the complete four-stage pipeline from input prompt to validated workflow code, illustrating how each training pair is constructed and verified.

9.2 LoRA Configuration↩︎

To mitigate the risk of mode collapse when fine-tuning on this limited dataset of approximately 1,300 examples, we employ Low-Rank Adaptation (LoRA) with rank-\(16\) [32] and train for one epoch with batch size of \(16\), as illustrated in Figure 4. This parameter-efficient approach serves as an effective regularization mechanism: preliminary experiments with full-parameter fine-tuning resulted in degenerate repetition patterns (Listing [lst:sft_mode_collapse]), where the model generates circular, non-terminating reasoning loops instead of producing valid workflows.

To solve this problem, we need to first understand what the problem is asking. The problem is asking us to find the answer. To find the answer, we need to solve this problem. To solve this problem, we need to first understand what the problem is asking. The problem is asking us to find the answer. To find the answer, we need to solve this problem. To solve this problem, we need to first understand what the problem is asking. The problem is asking us to find the answer. To find the answer, we need to solve this problem...

10 Details of Reinforcement Learning with Verifiable Rewards↩︎

The task set remains the same as the SFT stage: \[\mathcal{D}_{\text{train}}^{\text{RLVR}}=\mathcal{D}_{\text{train}}^{\text{SFT}}=\{\text{GSM8K, DROP, MBPP, Humaneval}\}\] To enhance the model’s generalization to diverse operator configurations, we train with four different operator sets that progressively introduce novel operators beyond the base SFT set:

\[\begin{align} \mathsf{Ops}^{\text{RLVR},1} &= (\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble}) \\ \mathsf{Ops}^{\text{RLVR},2} &= (\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble}, \texttt{Programmer}) \\ \mathsf{Ops}^{\text{RLVR},3} &= (\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble}, \texttt{Decompose}) \\ \mathsf{Ops}^{\text{RLVR},4} &= (\texttt{Generate}, \texttt{Summarize}, \texttt{Revise}, \texttt{Ensemble}, \texttt{Programmer}, \texttt{Decompose}) \end{align}\]

During each training iteration, we randomly sample one operator set from \(\{\mathsf{Ops}^{\text{RLVR},i}\}_{i=1}^4\) along with a task from \(\mathcal{D}_{\text{train}}^{\text{RLVR}}\), forming diverse domain combinations \((\mathsf{C}, \mathsf{Ops})\) for meta-learning. This diversity ensures the planner learns operator-agnostic workflow construction principles rather than memorizing fixed operator-task associations.

11 Case Study: Workflow Design Transformation After Training↩︎

11.1 Overview↩︎

To better understand the effectiveness of our training paradigm, we conduct a critical ablation study analyzing changes in model behavior before and after training. We compare two randomly selected workflows generated by the untrained base model (Qwen3-8B) versus the fully trained MetaFlow model after our two-stage training.

11.2 Impact of Training on Workflow Generation Behavior↩︎

Before Training (Base Model Qwen3-8B): Without SFT and RLVR optimization, the base model generates workflows with multi-step logic following a linear process: extract function name \(\rightarrow\) summarize \(\rightarrow\) generate \(\rightarrow\) revise \(\rightarrow\) ensemble.

class Workflow:
    # ... (initialization code omitted) ...
    async def run_workflow(self):
        # Step 1: Extract function name from test cases
        func_name = await self.generate(
            instruction="Extract the function name from the test cases. 
                        Return ONLY the function name, nothing else.",
            context=self.problem_text
        )
        # Step 2: Summarize problem description and test cases
        problem_summary = await self.summarize(
            instruction="Summarize the problem description and test cases 
                        to create a clear problem statement.",
            context=self.problem_text
        )
        # Step 3: Generate code based on problem summary
        generated_code = await self.generate(
            instruction=f"Write a Python function named '{func_name}'... 
                         Return ONLY the function definition, nothing else.",
            context=problem_summary
        )
        # Steps 4 & 5: Revise and ensemble
        revised_code = await self.revise(...)
        final_code = await self.ensemble(contexts=[generated_code, revised_code])
        return final_code

Analysis: While structured, this workflow has potential weaknesses:

1. Information Bottleneck: The summarize step, intended to condense information, risks losing critical details or constraints from the original problem description.

2. Instruction Ambiguity: Vague instructions like "Return ONLY the function definition" can be misinterpreted, leading to incomplete code that lacks necessary import statements or helper functions.

After Training (MetaFlow** Model):** In contrast, the MetaFlow model optimized through our two-stage training paradigm generates a workflow that, while maintaining complexity, is significantly more robust and methodical in its approach, and has learned the self-consistency operator unseen during training.

class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)

        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)
        self.selfconsistency = operator.SelfConsistency(self.llm, self.problem_text)

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        Remember:
        - Use detailed, comprehensive instructions
        - Dynamic instruction construction is powerful
        - All operators expect (instruction: str, context: str) except Ensemble which takes contexts: List[str]
        """
        import asyncio

        # Step 1: Extract function name from test cases
        function_name = await self.generate(
            instruction="""
            Analyze the test cases provided and extract the function name.
            The function name should match the naming convention used in the test cases.
            Return ONLY the function name, nothing else.
            """,
            context=self.problem_text
        )

        # Step 2: Extract input types and expected outputs from test cases
        input_types_output = await self.generate(
            instruction="""
            Analyze the test cases provided and extract the input types and expected outputs.
            Format the response as a dictionary where keys are the parameter names and values are the types.
            Also, identify the expected output format.
            Return ONLY the structured information, nothing else.
            """,
            context=self.problem_text
        )

        # Step 3: Generate multiple candidate implementations
        implementations = await asyncio.gather(
            self.generate(
                instruction=f"""
                Given the problem description: {self.problem_text}
                Generate a Python function that satisfies the test cases.
                Ensure that the function name is '{function_name}'.
                Include all necessary imports at the beginning of the function.
                Use the extracted input types and expected outputs: {input_types_output}.
                """,
                context=self.problem_text
            ),
            # ... (additional parallel generate calls omitted for brevity) ...
            self.generate(
                instruction=f"""
                Given the problem description: {self.problem_text}
                Generate another Python function that satisfies the test cases.
                Ensure that the function name is '{function_name}'.
                Include all necessary imports at the beginning of the function.
                Use the extracted input types and expected outputs: {input_types_output}.
                """,
                context=self.problem_text
            )
        )

        # Step 4: Select the best implementation using SelfConsistency
        best_implementation = await self.selfconsistency(
            instruction=f"""
            Evaluate the following candidate implementations:
            {implementations[0]}
            {implementations[1]}

            Select the best implementation based on the following criteria:
            - Consistency with the test cases
            - Logical correctness
            - Efficient and clean code
            - Proper use of imports and function signature

            Return ONLY the best implementation, nothing else.
            """,
            context=self.problem_text
        )

        return best_implementation

Analysis: The trained model’s workflow exhibits several key improvements contributing to its robustness:

1. From Compression to Structured Extraction: The model avoids the potentially lossy summarize step. Instead, it performs targeted extractions to parse unstructured text into structured data (e.g., function name, I/O types). This approach preserves information integrity and provides a more reliable foundation for subsequent steps.

2. Parallel Exploration for Increased Robustness: Rather than following a single generation-and-revision path, the model generates multiple, independent candidate solutions in parallel using asyncio.gather. This multi-path exploration inherently increases the robustness of the process by diversifying the solution space and reducing the risk of settling on a single, flawed implementation.

3. More Explicit and Context-Aware Instructions: The instructions are dynamically constructed using f-strings to include context from prior steps (e.g., ‘function_name’, ‘input_types_output’). This makes the prompts highly specific and unambiguous, directly mitigating the weaknesses of the base model and ensuring that each generative step is precisely guided.

12 Self Consistency Operator: Low-Cost Integration Example↩︎

This section demonstrates MetaFlow’s ability to immediately generate effective workflows with newly integrated operators without retraining. After introducing the SelfConsistency operator (which encapsulates parallel generation with majority voting), the model generates the following workflow for GSM8K mathematical reasoning through a single inference.

# --- DO NOT IMPORT HERE ---
class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)
        
        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)
        self.self_consistency = operator.SelfConsistency(
            self.llm, 
            self.problem_text,
            n_samples=5,                
            similarity_threshold=0.9,   
            return_full_info=False     
        )

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        Remember:
        - Use detailed, comprehensive instructions
        - Dynamic instruction construction is powerful
        - All operators expect (instruction: str, context: str) except Ensemble which takes contexts: List[str]
        """
        import asyncio
        sc_instruction = (
            "Solve this math problem step by step. "
            "Show all calculations clearly. "
            "At the end, state your final numerical answer."
        )
        
        sc_answer = await self.self_consistency(
            instruction=sc_instruction,
            context="",
            answer_type="numeric"
        )
        
        format_instruction = (
            f"The calculated answer is: {sc_answer}\n\n"
            "Extract ONLY the final numerical answer. "
            "If it's a decimal, round to a reasonable precision. "
            "If the answer represents a count of items/people, it should be a whole number. "
            "Return ONLY the number, nothing else."
        )
        
        final_answer = await self.generate(
            instruction=format_instruction,
            context=sc_answer
        )
        
        return final_answer.strip()

13 Case Study: Zero-Shot Workflow Generation with Novel VectorSearch Operator↩︎

This case study demonstrates MetaFlow’s zero-shot generalization to the novel VectorSearch operator on HotpotQA multi-hop question answering. The operator was never encountered during SFT or RLVR training, yet the model successfully generates workflows that orchestrate complex retrieval-reasoning pipelines. Listing [lst:hotpotqa_best_workflow] shows the highest-scoring workflow (0.74 accuracy) among 100 generated candidates, which achieves 60.9% improvement over the CoT+RAG baseline (0.46 accuracy).

# --- DO NOT IMPORT HERE ---
class Workflow:
    def __init__(self, config, problem) -> None:
        # --- DO NOT MODIFY THIS SECTION ---
        self.config = config
        self.problem_text = problem
        self.llm = create(config)

        self.generate = operator.Generate(self.llm, self.problem_text)
        self.revise = operator.Revise(self.llm, self.problem_text)
        self.summarize = operator.Summarize(self.llm, self.problem_text)
        self.ensemble = operator.Ensemble(self.llm, self.problem_text)
        self.vector_search = operator.VectorSearch(self.llm, self.problem_text)

    async def run_workflow(self):
        """
        Implement the core problem-solving logic here.
        Remember:
        - Use detailed, comprehensive instructions
        - Dynamic instruction construction is powerful
        - All operators expect (instruction: str, context: str) except Ensemble which takes contexts: List[str]
        """
        import asyncio

        # Step 1: Extract key entities from the question
        extraction_instruction = (
        "Identify all named entities, key terms, and concepts in the question. "
        "Focus on entities that are likely to appear in the documents and are critical for answering the question."
        )
        question_entities = await self.generate(instruction=extraction_instruction, context=self.problem_text)

        # Step 2: First Vector Search to retrieve initial relevant documents
        first_search_instruction = (
        f"Use the extracted entities: {question_entities}. "
        "Perform a vector search to find documents that mention these entities. "
        "Prioritize documents that provide foundational information about the question."
        )
        first_retrieved_docs = await self.vector_search(instruction=first_search_instruction, context=self.problem_text, top_k=5)

        # Step 3: Analyze first retrieved documents to identify connections
        analysis_instruction = (
        f"Based on the retrieved documents: {first_retrieved_docs}. "
        "Identify any shared entities, relationships, or connections between the documents. "
        "Highlight how these connections can help answer the question."
        )
        connections_analysis = await self.generate(instruction=analysis_instruction, context=self.problem_text)

        # Step 4: Second Vector Search to retrieve documents connecting the entities
        second_search_instruction = (
        f"Use the connections identified: {connections_analysis}. "
        "Perform a vector search to find documents that explicitly connect the entities or provide additional context."
        )
        second_retrieved_docs = await self.vector_search(instruction=second_search_instruction, context=self.problem_text, top_k=5)

        # Step 5: Synthesize information from both document sets
        synthesis_instruction = (
        f"Combine the information from the first set: {first_retrieved_docs}, "
        f"and the second set: {second_retrieved_docs}. "
        "Extract the exact answer to the question based on the synthesized information. "
        "Ensure the answer is concise and factually accurate."
        )
        final_answer = await self.generate(instruction=synthesis_instruction, context=self.problem_text)

        return final_answer

Analysis and Performance↩︎

the successful integration and application of a new tool. In Step 2, the model dynamically constructs a search query from its initial analysis and invokes the vectorsearch operator, effectively performing active information retrieval. When evaluated on the HotpotQA downstream task, this approach achieved a 60% search accuracy. This result is significant as it confirms that our operator framework enables models to learn and effectively utilize unseen tools.

Note on Comparability with Baselines↩︎

It is crucial to highlight a fundamental difference between our evaluation and that of many previous works on HotpotQA. Our methodology requires the model to actively perform a search to find relevant information. In contrast, prior baselines are often provided with the ground-truth supporting documents as part of their input, thereby bypassing the challenging information retrieval step entirely. Because our system solves a more complete and realistic version of the task that includes an explicit search phase, a direct comparison of end-to-end accuracy with such baselines is not meaningful.

References↩︎