Can I assign different states to every subagent in ParallelAgent?

[mrcaidev]

Here is what I'm trying to do: I'm building a crawler agent, which is expected to retrieve a list of news first, and then crawl every news in parallel.

fetch_list_agent -> fetch_news_agent_1
                 -> fetch_news_agent_2
                 -> fetch_news_agent_3
                 -> ...

And I suppose I have to assign different urls to every fetch_news_agent, so each of them crawls a different news. How can I do that?

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: The standard ParallelAgent is designed to run a fixed set of different agents concurrently, with each agent starting from the same shared state. It does not natively support distributing items from a list to identical sub-agents. The recommended approach is to build a custom agent to orchestrate this fan-out/parallel processing pattern.

---

Hello! That's an excellent question about how to handle parallel data processing in ADK.

Understanding ParallelAgent

The ParallelAgent is a workflow agent that executes a pre-defined list of sub-agents concurrently. [2] All sub-agents within it operate in independent branches but have access to the same initial shared session state. [1] For example, you could have a ParallelAgent that runs a WeatherFetcher agent and a NewsFetcher agent at the same time. [1]

This design is for task-parallelism (running different tasks simultaneously) rather than data-parallelism (running the same task on different pieces of data). In your case, all fetch_news_agent instances in a ParallelAgent would receive the same full list of URLs, and there isn't a built-in mechanism to assign a different URL to each one automatically.

Recommended Solution: Create a Custom Orchestrator Agent

For the kind of dynamic fan-out workflow you're building, the most flexible and powerful solution is to create a custom agent. By inheriting from BaseAgent, you can define the exact orchestration logic you need. [4]

Here’s how you could structure your solution:

1. FetchListAgent: This agent runs first and saves the list of news URLs into the session state, for example, context.state['news_urls'] = ['url1', 'url2', 'url3'].

2. CustomCrawlerAgent: This is your new custom agent. Its _run_async_impl method would orchestrate the parallel crawling.

3. Workflow: Your top-level agent would be a SequentialAgent that runs these two agents in order: SequentialAgent(sub_agents=[fetch_list_agent, custom_crawler_agent]).

Here is a conceptual example of what CustomCrawlerAgent might look like:

import asyncio
from google.adk.agents import BaseAgent
from google.adk.runtime import InvocationContext

# A helper async function to crawl a single URL.
# In a real implementation, you'd use a library like httpx or aiohttp.
async def crawl_url(url: str) -> tuple[str, str]:
    print(f"Crawling {url}...")
    # Simulate network request
    await asyncio.sleep(1)
    return (url, f"Content of {url}")

class CustomCrawlerAgent(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext):
        # 1. Read the list of URLs from the state
        urls_to_crawl = ctx.state.get('news_urls', [])
        if not urls_to_crawl:
            print("No URLs to crawl.")
            return

        # 2. Create a list of concurrent crawling tasks
        tasks = [crawl_url(url) for url in urls_to_crawl]

        # 3. Run all tasks in parallel and gather results
        results = await asyncio.gather(*tasks)

        # 4. Process results and save them back to the state
        crawled_content = dict(results)
        ctx.state['crawled_content'] = crawled_content
        print("Finished crawling all URLs.")

        # You can also yield events here to report progress
        yield self._create_event(ctx, text=f"Crawled {len(results)} URLs.")

This approach gives you full control over the parallel execution and directly solves the problem of processing a dynamic list of items concurrently.

---

[1] Multi-Agent Systems in ADK
[2] Parallel agents
[4] Custom agents

[mrcaidev]

The problem is the crawl_url also needs to be powered by an agent. Does it mean that I have to start a fresh session for each fetch_news_agent?