agentic-search-async-rl

name	agentic-search-async-rl
title	Beyond Ten Turns - Long-Horizon Agentic Search with Asynchronous RL
version	0.0.2
engine	skillxiv-v0.0.2-claude-opus-4.6
license	MIT
url	https://arxiv.org/abs/2508.07976
keywords	["agent-search","reinforcement-learning","long-horizon","asynchronous","autonomous-qa-generation"]
description	Enables long-horizon agentic search extending beyond 100 tool calls through scalable asynchronous RL training with autonomous QA dataset synthesis.

Beyond Ten Turns: Long-Horizon Agentic Search with Asynchronous RL

Core Concept

ASearcher overcomes the limitation that traditional online RL restricts agents to roughly 10 interaction turns. By employing scalable asynchronous RL training, the system enables agents to conduct searches extending beyond 100 tool calls while maintaining training efficiency. The approach includes autonomous LLM-based synthesis of large-scale QA datasets without external dependencies.

Architecture Overview

• Asynchronous RL Framework: Fully asynchronous training enabling long-horizon search
• Extended Tool Call Sequence: Support for 100+ tool interactions per episode
• Autonomous QA Synthesis: LLM agents autonomously generate high-quality QA pairs
• Large-Scale Dataset Generation: Create comprehensive evaluation benchmarks
• Zero-Shot Transfer: Base models work at inference without external LLMs

Implementation Steps

Step 1: Design Asynchronous RL Infrastructure

Create efficient asynchronous training system:

# Pseudocode for asynchronous RL
class AsyncSearchEnvironment(gym.Env):
    def __init__(self, tools, max_steps=200):
        super().__init__()
        self.tools = tools
        self.max_steps = max_steps
        self.current_step = 0
        self.search_history = []

    def step(self, action):
        """
        Execute action and return observation.

        Args:
            action: Tool to call and parameters

        Returns:
            observation: Result from tool
            reward: Reward signal
            done: Whether episode finished
            info: Metadata
        """
        self.current_step += 1

        # Execute tool asynchronously
        tool_name = action['tool']
        tool_params = action['params']

        try:
            result = self._execute_tool_async(tool_name, tool_params)
        except Exception as e:
            result = {'error': str(e)}

        # Track in history
        self.search_history.append({
            'step': self.current_step,
            'action': action,
            'result': result
        })

        # Compute reward
        reward = self._compute_step_reward(result, action)

        # Check termination
        done = (self.current_step >= .max_steps)  ._should_terminate()

         result, reward, done, {: .current_step}

     ():
        
         asyncio

          ():
            tool = .tools[tool_name]
              tool.execute_async(**params)

        
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(async_tool_call())

         result

     ():
        
           result:
             -

        
        info_gain = ((result)) / 
        reward = (info_gain, )

         reward

     ():
        
        
         (.search_history) >= :
            last_results = .search_history[-:]
             (  r[]  r  last_results):
                 

         

     ():
        
        .current_step = 
        .search_history = []
         {}

Step 2: Implement Asynchronous Training Loop

Create distributed training infrastructure:

# Pseudocode for async RL training
class AsyncRLTrainer:
    def __init__(self, model, environment, num_workers=16):
        super().__init__()
        self.model = model
        self.environment = environment
        self.num_workers = num_workers
        self.replay_buffer = deque(maxlen=100000)

    def collect_experience_async(self):
        """
        Asynchronously collect experience from multiple workers.
        """
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor(max_workers=self.num_workers) as executor:
            futures = []

            for worker_id in range(self.num_workers):
                future = executor.submit(self._run_worker_episode, worker_id)
                futures.append(future)

            # Collect results as they complete
            for future in concurrent.futures.as_completed(futures):
                trajectory = future.result()
                self.replay_buffer.append(trajectory)

        return len(self.replay_buffer)

    def _run_worker_episode(self, worker_id):
        """
        Run single episode in worker.
        """
        env = copy.deepcopy(self.environment)
        obs = env.reset()

        trajectory = {
            : [],
            : [],
            : [],
            : []
        }

        done = 
        step_count = 

          done  step_count < :
            
             torch.no_grad():
                action_logits = .model(obs)
                action = torch.multinomial(F.softmax(action_logits, dim=-), )

            
            obs, reward, done, info = env.step(._action_to_dict(action))

            
            trajectory[].append(obs)
            trajectory[].append(action.item())
            trajectory[].append(reward)
            trajectory[].append(obs)

            step_count += 

        
        trajectory[] = ._compute_returns(trajectory[])

         trajectory

     ():
        
        returns = []
        G = 
         r  (rewards):
            G = r +  * G
            returns.insert(, G)

         returns

     ():
        
         (.replay_buffer) < batch_size:
             

        batch = random.sample(.replay_buffer, batch_size)

        optimizer = AdamW(.model.parameters(), lr=)

         trajectory  batch:
            states = torch.tensor(trajectory[])
            actions = torch.tensor(trajectory[])
            returns = torch.tensor(trajectory[])

            
            logits = .model(states)
            log_probs = F.log_softmax(logits, dim=-)

            
            action_log_probs = log_probs.gather(, actions.unsqueeze(-))
            loss = -(action_log_probs * returns.unsqueeze(-)).mean()

            
            optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(.model.parameters(), )
            optimizer.step()

         loss.item()

     ():
        
         iteration  (num_iterations):
            
            num_trajectories = .collect_experience_async()

            
             _  ():  
                loss = .train_on_batch()

             (iteration + ) %  == :
                ()

         .model

Step 3: Implement Autonomous QA Generation

Generate training questions autonomously:

# Pseudocode for autonomous QA synthesis
class AutonomousQAGenerator:
    def __init__(self, qa_generation_model, synthesizer_tools):
        super().__init__()
        self.qa_model = qa_generation_model
        self.tools = synthesizer_tools

    def generate_question(self, seed_topic):
        """
        Generate a new question based on seed topic.

        Args:
            seed_topic: Domain or topic for question

        Returns:
            question: Generated question
        """
        prompt = f"""Generate a challenging question about {seed_topic} that requires web search to answer.
        The question should be specific enough to require at least 3-5 search steps to answer."""

        with torch.no_grad():
            question = self.qa_model.generate(
                prompt,
                max_length=100,
                temperature=0.8
            )

        return question

    def answer_question_autonomously(self, question):
        """
        Autonomously search and answer a question.

        Args:
            question: Question to answer

        Returns:
            answer: Generated answer
            search_path: Search actions taken
        """
        search_path = []
        search_queries = self._decompose_question(question)

        collected_info = []

        for query in search_queries[:5]:  # Limit to 5 search steps
            # Search
            results = self.tools['search'](query)
            search_path.append({'action': 'search', : query})

            
             result  results[:]:
                :
                    page_content = .tools[](result[])
                    collected_info.append(page_content)
                    search_path.append({: , : result[]})
                :
                    

        
        answer = ._synthesize_answer(question, collected_info)

         answer, search_path

     ():
        
        prompt = 

         torch.no_grad():
            queries_text = .qa_model.generate(prompt, max_length=)

        queries = [q.strip()  q  queries_text.split()]
         queries

     ():
        
        context = .join(information)

        prompt = 

         torch.no_grad():
            answer = .qa_model.generate(prompt, max_length=)

         answer

     ():
        
        dataset = []

         topic  topics:
             _  (num_per_topic):
                :
                    
                    question = .generate_question(topic)

                    
                    answer, search_path = .answer_question_autonomously(question)

                    dataset.append({
                        : question,
                        : answer,
                        : search_path,
                        : topic
                    })

                 Exception  e:
                    ()

         dataset

Step 4: Implement Long-Horizon Search Evaluation

Evaluate agent performance on extended searches:

# Pseudocode for long-horizon evaluation
class LongHorizonSearchEvaluator:
    def __init__(self, agent, tools):
        super().__init__()
        self.agent = agent
        self.tools = tools

    def evaluate_agent(self, test_questions, max_steps=150):
        """
        Evaluate agent on test questions with long horizons.

        Args:
            test_questions: List of questions to answer
            max_steps: Maximum steps allowed

        Returns:
            results: Evaluation metrics
        """
        results = {
            'success_rate': 0,
            'avg_steps': 0,
            'answer_quality': 0,
            'long_horizon_success': 0
        }

        correct_answers = 0
        total_steps = 0
        long_horizon_attempts = 0

        for question in test_questions:
            # Initialize search
            obs = {'question': question, 'search_history': []}

            answer = None
            step_count = 0
            found_answer = False

            # Run search
            while step_count < max_steps:
                # Agent decides next action
                with torch.no_grad():
                    action = self.agent.decide_action(obs)

                if action['type'] == :
                    answer = action.get()
                    found_answer = 
                    

                
                 action[] == :
                    results_list = .tools[](action[])
                    obs[].append({
                        : ,
                        : action[],
                        : results_list
                    })

                 action[] == :
                    content = .tools[](action[])
                    obs[].append({
                        : ,
                        : action[],
                        : content
                    })

                step_count += 

             found_answer:
                correct_answers += 

            total_steps += step_count

            
             step_count > :
                long_horizon_attempts += 

        
        results[] = correct_answers / (test_questions)
        results[] = total_steps / (test_questions)
        results[] = long_horizon_attempts

         results

Practical Guidance

Hyperparameters and Configuration:

• Number of async workers: 8-32 depending on infrastructure
• Maximum steps per episode: 100-200
• Replay buffer size: 50k-200k trajectories
• RL learning rate: 1e-5 to 5e-5
• Training iterations: 500-2000

When to Use Long-Horizon Agentic Search:

• Complex multi-step research and information retrieval
• Scenarios requiring iterative refinement and exploration
• Systems where agent must autonomously discover solution paths
• Applications with 30+ step search horizons

When NOT to Use:

• Simple factual lookup (can be answered in <5 steps)
• Scenarios with unreliable or sparse information sources
• Real-time systems with strict step budgets
• When deterministic paths are more reliable than learned policies

Implementation Notes:

• Asynchronous training critical for handling long episodes
• QA dataset generation should maintain diversity across domains
• Monitor success rates at different horizon lengths (10, 50, 100+ steps)
• Consider curriculum learning: start with short horizons, gradually increase
• Autonomous QA generation requires periodic manual validation

Reference

Paper: Beyond Ten Turns: Long-Horizon Agentic Search with Asynchronous RL ArXiv: 2508.07976 Performance: Enables 100+ tool calls and 400,000+ output tokens during training, enables agents to develop sophisticated multi-step search strategies