ADu2021/agentic-search-async-rl

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 >= self.max_steps) or self._should_terminate()

        return result, reward, done, {'step': self.current_step}

    def _execute_tool_async(self, tool_name, params):
        """
        Execute tool asynchronously to enable long horizons.
        """
        import asyncio

        async def async_tool_call():
            tool = self.tools[tool_name]
            return await tool.execute_async(**params)

        # Use event loop for async execution
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(async_tool_call())

        return result

    def _compute_step_reward(self, result, action):
        """
        Compute reward for this step.
        """
        if 'error' in result:
            return -0.1

        # Reward for useful information
        info_gain = len(str(result)) / 1000.0
        reward = min(info_gain, 0.5)

        return reward

    def _should_terminate(self):
        """
        Decide if search should terminate early.
        """
        # Check if solution found
        if len(self.search_history) >= 3:
            last_results = self.search_history[-3:]
            if all('answer' in r['result'] for r in last_results):
                return True

        return False

    def reset(self):
        """
        Reset environment for new episode.
        """
        self.current_step = 0
        self.search_history = []
        return {}

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 = {
            'states': [],
            'actions': [],
            'rewards': [],
            'observations': []
        }

        done = False
        step_count = 0

        while not done and step_count < 200:
            # Agent selects action
            with torch.no_grad():
                action_logits = self.model(obs)
                action = torch.multinomial(F.softmax(action_logits, dim=-1), 1)

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

            # Store trajectory
            trajectory['states'].append(obs)
            trajectory['actions'].append(action.item())
            trajectory['rewards'].append(reward)
            trajectory['observations'].append(obs)

            step_count += 1

        # Compute returns
        trajectory['returns'] = self._compute_returns(trajectory['rewards'])

        return trajectory

    def _compute_returns(self, rewards):
        """
        Compute discounted returns.
        """
        returns = []
        G = 0
        for r in reversed(rewards):
            G = r + 0.99 * G
            returns.insert(0, G)

        return returns

    def train_on_batch(self, batch_size=128):
        """
        Train on batch from replay buffer.
        """
        if len(self.replay_buffer) < batch_size:
            return None

        batch = random.sample(self.replay_buffer, batch_size)

        optimizer = AdamW(self.model.parameters(), lr=1e-5)

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

            # Forward pass
            logits = self.model(states)
            log_probs = F.log_softmax(logits, dim=-1)

            # Policy gradient loss
            action_log_probs = log_probs.gather(1, actions.unsqueeze(-1))
            loss = -(action_log_probs * returns.unsqueeze(-1)).mean()

            # Backward pass
            optimizer.zero_grad()
            loss.backward()
            torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
            optimizer.step()

        return loss.item()

    def train(self, num_iterations=1000):
        """
        Full async RL training loop.
        """
        for iteration in range(num_iterations):
            # Collect experience asynchronously
            num_trajectories = self.collect_experience_async()

            # Train on batch
            for _ in range(10):  # Multiple training iterations per collection
                loss = self.train_on_batch()

            if (iteration + 1) % 10 == 0:
                print(f"Iteration {iteration+1}: Buffer size = {num_trajectories}")

        return self.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': query})

            # Read relevant pages
            for result in results[:2]:
                try:
                    page_content = self.tools['read_page'](result['url'])
                    collected_info.append(page_content)
                    search_path.append({'action': 'read', 'url': result['url']})
                except:
                    pass

        # Synthesize answer
        answer = self._synthesize_answer(question, collected_info)

        return answer, search_path

    def _decompose_question(self, question):
        """
        Break question into search queries.
        """
        prompt = f"""Break down this question into 3-5 search queries:
        {question}

        Return as comma-separated list."""

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

        queries = [q.strip() for q in queries_text.split(',')]
        return queries

    def _synthesize_answer(self, question, information):
        """
        Generate answer from collected information.
        """
        context = '\n'.join(information)

        prompt = f"""Question: {question}

Information collected:
{context}

Please provide a clear, concise answer based on the information above."""

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

        return answer

    def generate_qa_dataset(self, topics, num_per_topic=10):
        """
        Generate large-scale QA dataset autonomously.

        Args:
            topics: List of seed topics
            num_per_topic: Questions to generate per topic

        Returns:
            dataset: QA dataset
        """
        dataset = []

        for topic in topics:
            for _ in range(num_per_topic):
                try:
                    # Generate question
                    question = self.generate_question(topic)

                    # Answer autonomously
                    answer, search_path = self.answer_question_autonomously(question)

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

                except Exception as e:
                    print(f"Error generating QA for {topic}: {e}")

        return 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'] == 'synthesize':
                    answer = action.get('answer')
                    found_answer = True
                    break

                # Execute action
                if action['type'] == 'search':
                    results_list = self.tools['search'](action['query'])
                    obs['search_history'].append({
                        'type': 'search',
                        'query': action['query'],
                        'results': results_list
                    })

                elif action['type'] == 'read_page':
                    content = self.tools['read_page'](action['url'])
                    obs['search_history'].append({
                        'type': 'read',
                        'url': action['url'],
                        'content': content
                    })

                step_count += 1

            if found_answer:
                correct_answers += 1

            total_steps += step_count

            # Track long-horizon success (>30 steps)
            if step_count > 30:
                long_horizon_attempts += 1

        # Compute metrics
        results['success_rate'] = correct_answers / len(test_questions)
        results['avg_steps'] = total_steps / len(test_questions)
        results['long_horizon_success'] = long_horizon_attempts

        return 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