The Pattern
Create specialized agents for different tasks, then orchestrate them with a coordinator: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: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.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.
AgentServer, this wiring is automatic:
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: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
Keep agents focused
Keep agents focused
Each agent should have a clear, specific role. Don’t overload them.
Use different prompting styles
Use different prompting styles
Customize system prompts for each agent’s specialty and personality.
Consider using different models
Consider using different models
Use faster/cheaper models for simple tasks, more capable ones for complex reasoning.
Add error handling
Add error handling
Wrap agent calls in try/catch and have fallback strategies.