> ## Documentation Index > Fetch the complete documentation index at: https://thethirdpenco-feat-tool-lifecycle-events.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # Multi-Agent Workflow > Coordinate multiple specialized agents to complete complex tasks Build workflows where multiple specialized agents collaborate to complete complex tasks. ## The Pattern Create specialized agents for different tasks, then orchestrate them with a coordinator: ```python theme={null} from ai_query import generate_text from ai_query.providers import google, tool, Field # Specialized "agents" as tools @tool(description="Research agent: searches and gathers information") async def researcher(topic: str = Field(description="Topic to research")) -> str: result = await generate_text( model=google("gemini-2.0-flash"), system="You are a research specialist. Provide detailed findings.", prompt=f"Research: {topic}", ) return result.text @tool(description="Writer agent: creates content from research") async def writer( topic: str = Field(description="Topic"), research: str = Field(description="Research findings to use"), ) -> str: result = await generate_text( model=google("gemini-2.0-flash"), system="You are a content writer. Create engaging content.", prompt=f"Write about {topic} using this research:\n{research}", ) return result.text # Coordinator orchestrates the workflow result = await generate_text( model=google("gemini-2.0-flash"), system="""You are a project coordinator. To complete tasks: 1. Use the researcher to gather information 2. Use the writer to create the final content 3. Return the final result""", prompt="Create a blog post about renewable energy trends in 2024", tools={"researcher": researcher, "writer": writer}, ) ``` ## Complete Example: Content Pipeline A full content creation pipeline with research, writing, and editing: ```python theme={null} import asyncio from ai_query import generate_text from ai_query.providers import google, tool, Field, step_count_is class ContentPipeline: def __init__(self): self.research_data = None self.draft = None self.final = None async def create_content(self, topic: str) -> str: @tool(description="Research agent: gathers comprehensive information on a topic") async def research( query: str = Field(description="What to research"), ) -> str: result = await generate_text( model=google("gemini-2.0-flash"), system="""You are a thorough researcher. Provide: - Key facts and statistics - Recent developments - Expert opinions - Relevant examples""", prompt=query, ) self.research_data = result.text return f"Research complete. Key findings:\n{result.text}" @tool(description="Writer agent: creates a well-structured draft") async def write_draft( topic: str = Field(description="Topic to write about"), research: str = Field(description="Research to incorporate"), ) -> str: result = await generate_text( model=google("gemini-2.0-flash"), system="""You are a skilled writer. Create engaging, well-structured content with clear sections.""", prompt=f"Write about: {topic}\n\nUsing research:\n{research}", ) self.draft = result.text return f"Draft created:\n{result.text}" @tool(description="Editor agent: refines and polishes the draft") async def edit( draft: str = Field(description="Draft to edit"), ) -> str: result = await generate_text( model=google("gemini-2.0-flash"), system="""You are an editor. Improve the text by: - Fixing any errors - Improving clarity and flow - Strengthening the opening and closing - Adding transitions between sections""", prompt=f"Edit this draft:\n{draft}", ) self.final = result.text return f"Editing complete:\n{result.text}" @tool(description="Mark the content as complete and ready for publication") def complete(final_content: str = Field(description="Final content")) -> str: self.final = final_content return "Content pipeline complete!" await generate_text( model=google("gemini-2.0-flash"), system="""You are a content pipeline coordinator. Follow this workflow: 1. Research the topic thoroughly 2. Write a draft using the research 3. Edit and polish the draft 4. Call complete with the final content Each agent specializes in their task. Use them in order.""", prompt=f"Create high-quality content about: {topic}", tools={ "research": research, "write_draft": write_draft, "edit": edit, "complete": complete, }, stop_when=step_count_is(6), ) return self.final async def main(): pipeline = ContentPipeline() content = await pipeline.create_content( "The impact of AI on software development in 2024" ) print(content) asyncio.run(main()) ``` ## Distributed Agents (Registry) You can build systems where some agents are local and others are remote (Serverless, Cloud Run, etc.). The **Agent Registry** makes this transparent to your code. ```python theme={null} from ai_query import AgentRegistry, Agent, HTTPTransport from ai_query.providers import openai # 1. Define Local Agents class Coordinator(Agent): async def run_workflow(self, topic: str): # Call 'researcher' by ID # The registry decides if it's local or remote research = await self.call("researcher").research(topic) return research # 2. Configure Registry registry = AgentRegistry() # Map 'coordinator' to local class registry.register("coordinator", Coordinator) # Map 'researcher' to remote serverless function registry.register( "researcher", HTTPTransport("https://api.example.com/agent/researcher") ) # 3. Run coordinator = Coordinator("coordinator") # Inject registry-aware transport (handled by AgentServer in production) # For standalone script usage, we can manually wire it or use AgentServer. ``` When using `AgentServer`, this wiring is automatic: ```python theme={null} from ai_query import AgentServer, AgentRegistry, HTTPTransport registry = AgentRegistry() registry.register("coordinator", CoordinatorAgent) registry.register("worker-.*", HTTPTransport("https://worker-cluster.com/agent")) # Server handles routing between local and remote agents server = AgentServer(registry) server.serve() ``` ## Using Stateful Agents For more complex workflows where agents need to maintain state: ```python theme={null} from ai_query.agents import Agent, MemoryStorage from ai_query.providers import openai class ResearcherAgent(Agent): def __init__(self): super().__init__( "researcher", model=openai("gpt-4o"), system="You are a thorough researcher. Gather comprehensive information.", storage=MemoryStorage(), initial_state={"findings": []} ) async def research(self, topic: str) -> str: response = await self.chat(f"Research: {topic}") await self.update_state( findings=self.state["findings"] + [{"topic": topic, "result": response}] ) return response class WriterAgent(Agent): def __init__(self): super().__init__( "writer", model=openai("gpt-4o"), system="You are a skilled content writer.", storage=MemoryStorage() ) async def write(self, topic: str, research: str) -> str: return await self.chat(f"Write about {topic} using:\n{research}") # Orchestrate the workflow async def create_content(topic: str) -> str: researcher = ResearcherAgent() writer = WriterAgent() async with researcher, writer: # Step 1: Research research = await researcher.research(topic) # Step 2: Write content = await writer.write(topic, research) return content ``` ## Tips Each agent should have a clear, specific role. Don't overload them. Customize system prompts for each agent's specialty and personality. Use faster/cheaper models for simple tasks, more capable ones for complex reasoning. Wrap agent calls in try/catch and have fallback strategies.