The AI One-Person Company Playbook: Tools, Agents, and Automation
The Rise of the AI One-Person Company
The AI one-person company represents a fundamental shift in how software businesses are built. Instead of hiring teams of engineers, designers, and operations staff, a single developer leverages large language models, autonomous agents, and workflow automation to multiply their output 10x or more. You become the orchestrator of an AI workforce rather than a solo builder doing everything manually.
This playbook covers the concrete tools, agent architectures, and automation patterns that make this possible today ā not in some hypothetical future, but with technology you can deploy right now.
What Exactly Is an AI One-Person Company?
At its core, an AI one-person company is a business where a single human operator uses AI systems to handle the work that would traditionally require multiple employees. The key distinction isn't just using AI as a coding assistant ā it's structuring your entire operation around AI agents that operate semi-autonomously, making decisions, executing tasks, and communicating results back to you.
A typical stack looks like this:
Orchestration Layer ā You, the human, defining goals, reviewing output, and making high-level decisions
Agent Layer ā Specialized AI agents that handle distinct domains (coding, customer support, content creation, data analysis)
Tool Layer ā APIs, databases, and external services that agents interact with
Automation Layer ā Scheduled jobs, webhooks, and event-driven workflows that keep everything running
Memory/Context Layer ā Vector databases, conversation logs, and knowledge bases that give agents long-term memory
Why This Matters Now
The economics have shifted dramatically. A skilled developer can now build and maintain a SaaS product, handle customer support, create marketing content, and manage operations ā all with AI assistance. The cost structure of a one-person AI company is fundamentally different from a traditional startup: your burn rate stays near zero while your capability ceiling keeps rising as models improve.
Three factors converged to make this possible:
Model capability leap ā GPT-4, Claude, and open-source models can now reason, plan, and execute complex multi-step tasks
Tool integration maturity ā Function calling, structured outputs, and API ecosystems let agents interact with real-world systems reliably
Orchestration frameworks ā Libraries like LangChain, CrewAI, and AutoGen provide production-ready patterns for multi-agent coordination
The Core Architecture: Building Your AI Workforce
Let's build a concrete example. Imagine you're running a SaaS product that needs: bug fixes and feature development, customer support email handling, and social media content creation. Here's how you'd structure the agent system.
The Orchestrator Pattern
The orchestrator is a central agent that decomposes high-level tasks and delegates them to specialized sub-agents. It maintains context across interactions and synthesizes results.
import os
from typing import Dict, List, Optional
from dataclasses import dataclass, field
import json
from openai import OpenAI
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
@dataclass
class Task:
"""A decomposable unit of work for the agent system."""
id: str
description: str
agent_type: str # 'coder', 'support', 'content', 'orchestrator'
dependencies: List[str] = field(default_factory=list)
status: str = "pending"
result: Optional[str] = None
class OrchestratorAgent:
"""
Central agent that breaks down complex goals into subtasks,
routes them to specialized agents, and aggregates results.
"""
def __init__(self):
self.tasks: Dict[str, Task] = {}
self.completed_tasks: List[Task] = []
self.agent_registry = {
"coder": CoderAgent(),
"support": SupportAgent(),
"content": ContentAgent(),
}
def decompose_goal(self, goal: str) -> List[Task]:
"""Use LLM to break a high-level goal into concrete tasks."""
prompt = f"""You are a technical project manager. Given this goal:
"{goal}"
Break it down into 3-7 concrete, executable tasks. For each task, specify:
1. A unique task ID (e.g., T1, T2)
2. A clear description
3. The agent type needed: 'coder' (for code/technical work),
'support' (for customer-facing work), or 'content' (for writing/marketing)
4. Any dependencies (list of task IDs that must complete first)
Output as JSON array of tasks. Example format:
[{{"id": "T1", "description": "Set up database schema", "agent_type": "coder", "dependencies": []}}]"""
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "system", "content": prompt}],
response_format={"type": "json_object"}
)
tasks_data = json.loads(response.choices[0].message.content)
tasks = []
for t in tasks_data.get("tasks", []):
task = Task(
id=t["id"],
description=t["description"],
agent_type=t["agent_type"],
dependencies=t.get("dependencies", [])
)
self.tasks[task.id] = task
tasks.append(task)
return tasks
def execute_task(self, task: Task) -> str:
"""Route a task to the appropriate agent and capture the result."""
agent = self.agent_registry.get(task.agent_type)
if not agent:
return f"No agent available for type: {task.agent_type}"
print(f"š Dispatching {task.id} to {task.agent_type} agent...")
result = agent.execute(task.description)
task.status = "completed"
task.result = result
self.completed_tasks.append(task)
return result
def run_workflow(self, goal: str) -> str:
"""Execute a full workflow from goal to completion."""
tasks = self.decompose_goal(goal)
print(f"š Decomposed goal into {len(tasks)} tasks")
completed_ids = set()
results_log = []
while len(completed_ids) < len(tasks):
for task in tasks:
if task.id in completed_ids:
continue
if all(dep in completed_ids for dep in task.dependencies):
result = self.execute_task(task)
completed_ids.add(task.id)
results_log.append(f"[{task.id}] {task.description}: {result[:200]}")
# Synthesize final summary
summary_prompt = f"""Synthesize these completed task results into a concise status report:
Goal: {goal}
Results:
{chr(10).join(results_log)}
Provide a clear summary of what was accomplished and any next steps."""
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[{"role": "user", "content": summary_prompt}]
)
return response.choices[0].message.content
Specialized Agent Implementation
Each specialized agent has its own system prompt, tool set, and execution logic. Here's how to build them:
class CoderAgent:
"""Agent specialized in software development tasks."""
def __init__(self):
self.system_prompt = """You are an expert software engineer.
You write clean, well-documented, production-ready code.
When given a task, you:
1. Analyze requirements carefully
2. Plan the implementation approach
3. Write complete, working code with error handling
4. Include tests where appropriate
5. Explain any trade-offs or assumptions"""
def execute(self, task_description: str) -> str:
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"Complete this coding task:\n{task_description}\n\nProvide the full implementation with explanations."}
],
temperature=0.3 # Lower temperature for coding precision
)
return response.choices[0].message.content
class SupportAgent:
"""Agent specialized in customer support and communication."""
def __init__(self):
self.system_prompt = """You are a helpful, empathetic customer support specialist.
Your approach:
1. Acknowledge the customer's concern with empathy
2. Diagnose the issue by asking clarifying questions if needed
3. Provide a clear, actionable solution
4. Confirm resolution and offer further assistance
5. Maintain a professional, warm tone throughout"""
def execute(self, task_description: str) -> str:
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"Handle this customer support scenario:\n{task_description}"}
],
temperature=0.7 # Higher temperature for natural conversation
)
return response.choices[0].message.content
class ContentAgent:
"""Agent specialized in content creation and marketing."""
def __init__(self):
self.system_prompt = """You are a skilled content marketer and writer.
You create engaging, SEO-friendly content that drives results.
Your process:
1. Understand the target audience and platform
2. Craft compelling headlines and hooks
3. Write clear, valuable content with appropriate tone
4. Include calls-to-action where relevant
5. Optimize for the specified platform (Twitter, blog, email, etc.)"""
def execute(self, task_description: str) -> str:
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": f"Create content for this requirement:\n{task_description}"}
],
temperature=0.8 # Creative tasks benefit from higher temperature
)
return response.choices[0].message.content
Agents become truly useful when they can interact with external systems ā databases, APIs, file systems, and communication channels. This is where function calling transforms an AI assistant into an autonomous operator.
Defining a Tool Registry
import subprocess
import requests
from typing import Callable, Any
import sqlite3
from datetime import datetime
class ToolRegistry:
"""Registry of functions that agents can call to interact with the world."""
def __init__(self):
self.tools: Dict[str, dict] = {}
def register(self, name: str, description: str, func: Callable, parameters: dict):
"""Register a callable tool with its schema."""
self.tools[name] = {
"name": name,
"description": description,
"function": func,
"parameters": parameters
}
def get_openai_schema(self) -> List[dict]:
"""Convert registered tools to OpenAI function calling format."""
return [
{
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t["parameters"]
}
}
for t in self.tools.values()
]
def execute(self, tool_name: str, arguments: dict) -> str:
"""Execute a registered tool with the given arguments."""
tool = self.tools.get(tool_name)
if not tool:
return f"Error: Tool '{tool_name}' not found"
try:
result = tool["function"](**arguments)
return str(result)
except Exception as e:
return f"Error executing {tool_name}: {str(e)}"
# Initialize the tool registry
tool_registry = ToolRegistry()
# Register a database query tool
tool_registry.register(
name="query_database",
description="Execute a SQL query against the application database",
func=lambda query: _execute_db_query(query),
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The SQL query to execute (SELECT only for safety)"
}
},
"required": ["query"]
}
)
# Register a deployment tool
tool_registry.register(
name="deploy_to_production",
description="Deploy the latest version of the application to production",
func=lambda environment, version: _deploy_app(environment, version),
parameters={
"type": "object",
"properties": {
"environment": {
"type": "string",
"enum": ["staging", "production"],
"description": "Target deployment environment"
},
"version": {
"type": "string",
"description": "Version tag to deploy"
}
},
"required": ["environment", "version"]
}
)
# Register a send email tool
tool_registry.register(
name="send_email",
description="Send an email to a customer or team member",
func=lambda to, subject, body: _send_email(to, subject, body),
parameters={
"type": "object",
"properties": {
"to": {"type": "string", "description": "Recipient email address"},
"subject": {"type": "string", "description": "Email subject line"},
"body": {"type": "string", "description": "Email body content"}
},
"required": ["to", "subject", "body"]
}
)
# Register a web scraper tool
tool_registry.register(
name="fetch_webpage",
description="Fetch and extract text content from a URL",
func=lambda url: _fetch_webpage_content(url),
parameters={
"type": "object",
"properties": {
"url": {"type": "string", "description": "URL to fetch content from"}
},
"required": ["url"]
}
)
def _execute_db_query(query: str) -> str:
"""Internal function to safely execute database queries."""
conn = sqlite3.connect("app_database.db")
cursor = conn.cursor()
try:
cursor.execute(query)
results = cursor.fetchall()
return json.dumps(results, default=str)
finally:
conn.close()
def _deploy_app(environment: str, version: str) -> str:
"""Internal deployment function."""
result = subprocess.run(
["deploy-script.sh", "--env", environment, "--version", version],
capture_output=True, text=True
)
return f"Deployment to {environment} (version {version}):\n{result.stdout}"
def _send_email(to: str, subject: str, body: str) -> str:
"""Send email via configured provider."""
# Integration with SendGrid, Mailgun, or SMTP
response = requests.post(
"https://api.sendgrid.com/v3/mail/send",
headers={"Authorization": f"Bearer {os.environ['SENDGRID_API_KEY']}"},
json={
"personalizations": [{"to": [{"email": to}]}],
"subject": subject,
"content": [{"type": "text/plain", "value": body}]
}
)
return f"Email sent to {to}: {response.status_code}"
def _fetch_webpage_content(url: str) -> str:
"""Fetch and parse webpage content."""
response = requests.get(url, headers={"User-Agent": "AI-Agent/1.0"})
# Strip HTML tags for simplicity ā in production use BeautifulSoup
import re
text = re.sub(r'<[^>]+>', ' ', response.text)
return text[:5000] # Truncate to reasonable context window
Agent with Tool Access
class ToolEnabledAgent:
"""An agent that can autonomously decide when to use tools."""
def __init__(self, system_prompt: str, tool_registry: ToolRegistry):
self.system_prompt = system_prompt
self.tool_registry = tool_registry
self.conversation_history = []
def execute_with_tools(self, user_request: str, max_tool_calls: int = 5) -> str:
"""Process a request, allowing the agent to call tools as needed."""
self.conversation_history = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_request}
]
tool_calls_made = 0
while tool_calls_made < max_tool_calls:
response = client.chat.completions.create(
model="gpt-4-turbo-preview",
messages=self.conversation_history,
tools=self.tool_registry.get_openai_schema(),
tool_choice="auto"
)
message = response.choices[0].message
# If no tool calls, the agent is done
if not message.tool_calls:
self.conversation_history.append({
"role": "assistant",
"content": message.content
})
return message.content
# Process tool calls
self.conversation_history.append({
"role": "assistant",
"content": message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in message.tool_calls
]
})
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
print(f"š§ Agent calling tool: {tool_name}({arguments})")
result = self.tool_registry.execute(tool_name, arguments)
self.conversation_history.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
tool_calls_made += 1
return "Agent reached maximum tool call limit without completing the task."
Automation: The Always-On Layer
Your AI workforce needs scheduled jobs and event-driven triggers to operate continuously. This automation layer handles recurring tasks without your intervention.
Scheduled Task Runner
import schedule
import time
from datetime import datetime
import threading
class AutomationEngine:
"""Runs scheduled and event-driven tasks using the agent workforce."""
def __init__(self, orchestrator: OrchestratorAgent):
self.orchestrator = orchestrator
self.jobs = []
self.running = False
self._thread = None
def schedule_daily_briefing(self, time_str: str = "09:00"):
"""Schedule a daily AI-generated briefing of priorities."""
def briefing_job():
print(f"\nš [{datetime.now()}] Running daily briefing...")
result = self.orchestrator.run_workflow(
"Analyze yesterday's customer support tickets, "
"identify top issues, prioritize bug fixes, "
"and suggest content for today's social media posts."
)
print(f"Briefing complete: {result[:300]}...")
# Save briefing to a log file
with open("daily_briefings.log", "a") as f:
f.write(f"\n=== Briefing {datetime.now().date()} ===\n{result}\n")
schedule.every().day.at(time_str).do(briefing_job)
self.jobs.append(("daily_briefing", time_str))
def schedule_code_review(self, interval_hours: int = 4):
"""Periodically review recent commits for quality issues."""
def review_job():
print(f"\nš [{datetime.now()}] Running automated code review...")
result = self.orchestrator.run_workflow(
"Check the latest git commits from the past 4 hours, "
"review for potential bugs, security issues, or code smells, "
"and create fix tickets for any issues found."
)
print(f"Code review complete")
schedule.every(interval_hours).hours.do(review_job)
self.jobs.append(("code_review", f"every_{interval_hours}h"))
def schedule_customer_support_digest(self):
"""Process support tickets every 30 minutes."""
def support_job():
print(f"\nš¬ [{datetime.now()}] Processing support queue...")
result = self.orchestrator.run_workflow(
"Check for new unanswered customer support emails, "
"draft responses for each, and flag any urgent issues "
"that require immediate human attention."
)
print(f"Support digest complete")
schedule.every(30).minutes.do(support_job)
self.jobs.append(("support_digest", "every_30min"))
def start(self):
"""Start the automation engine in a background thread."""
self.running = True
self._thread = threading.Thread(target=self._run_loop, daemon=True)
self._thread.start()
print("āļø Automation engine started")
def _run_loop(self):
while self.running:
schedule.run_pending()
time.sleep(1)
def stop(self):
self.running = False
print("Automation engine stopped")
def list_jobs(self):
return [(name, schedule_info) for name, schedule_info in self.jobs]
# Usage example
orchestrator = OrchestratorAgent()
automation = AutomationEngine(orchestrator)
automation.schedule_daily_briefing("09:00")
automation.schedule_customer_support_digest()
automation.schedule_code_review(4)
automation.start()
print("Your AI workforce is now running autonomously.")
print("Scheduled jobs:", automation.list_jobs())
# Keep the main thread alive
try:
while True:
time.sleep(60)
except KeyboardInterrupt:
automation.stop()
Webhook-Driven Agent Triggers
from flask import Flask, request, jsonify
import hashlib
import hmac
app = Flask(__name__)
class WebhookHandler:
"""Handles incoming webhooks and routes them to appropriate agents."""
def __init__(self, orchestrator: OrchestratorAgent):
self.orchestrator = orchestrator
self.handlers = {}
def register_handler(self, event_type: str, handler_func: Callable):
self.handlers[event_type] = handler_func
def process_webhook(self, source: str, event_type: str, payload: dict) -> str:
"""Process an incoming webhook event."""
if event_type in self.handlers:
return self.handlers[event_type](payload)
# Default: let the orchestrator figure it out
goal = f"Handle this {source} webhook event: {event_type}\nPayload: {json.dumps(payload)}"
return self.orchestrator.run_workflow(goal)
webhook_handler = WebhookHandler(orchestrator)
# Register specific handlers
def handle_stripe_payment_succeeded(payload: dict):
"""When a customer pays, update their account and send welcome email."""
customer_email = payload["data"]["object"]["customer_email"]
amount = payload["data"]["object"]["amount"] / 100
return orchestrator.run_workflow(
f"A customer just paid ${amount}. Their email is {customer_email}. "
f"Update their account status to 'active', send a thank-you email, "
f"and log the transaction in the database."
)
def handle_github_push(payload: dict):
"""On new code push, run tests and prepare deployment notes."""
repo_name = payload["repository"]["full_name"]
commit_msg = payload["head_commit"]["message"]
return orchestrator.run_workflow(
f"New commit pushed to {repo_name}: '{commit_msg}'. "
f"Run the test suite, check for breaking changes, "
f"and prepare release notes for the changes."
)
webhook_handler.register_handler("payment.succeeded", handle_stripe_payment_succeeded)
webhook_handler.register_handler("push", handle_github_push)
@app.route("/webhook/", methods=["POST"])
def receive_webhook(source):
"""Receive webhooks from external services."""
payload = request.json
event_type = request.headers.get("X-Event-Type", "unknown")
# Verify webhook signature (example for Stripe-style verification)
signature = request.headers.get("X-Signature")
if signature:
expected = hmac.new(
os.environ["WEBHOOK_SECRET"].encode(),
request.data,
hashlib.sha256
).hexdigest()
if not hmac.compare_digest(signature, expected):
return jsonify({"error": "Invalid signature"}), 401
print(f"šØ Received {source} webhook: {event_type}")
result = webhook_handler.process_webhook(source, event_type, payload)
return jsonify({"status": "processed", "summary": result[:500]})
Memory and Context: Giving Agents Long-Term Recall
Agents become exponentially more useful when they remember past interactions, decisions, and learned patterns. A vector database backed memory system is essential.
import chromadb
from chromadb.utils import embedding_functions
import uuid
class AgentMemory:
"""Persistent memory for agents using vector embeddings and metadata."""
def __init__(self, agent_name: str):
self.client = chromadb.PersistentClient(path="./agent_memory")
self.embedding_fn = embedding_functions.OpenAIEmbeddingFunction(
api_key=os.environ["OPENAI_API_KEY"],
model_name="text-embedding-3-small"
)
self.collection = self.client.get_or_create_collection(
name=f"agent_{agent_name}",
embedding_function=self.embedding_fn
)
self.agent_name = agent_name
def remember(self, content: str, metadata: dict = None):
"""Store a memory with optional metadata."""
memory_id = str(uuid.uuid4())
self.collection.add(
ids=[memory_id],
documents=[content],
metadatas=[metadata or {}]
)
return memory_id
def recall(self, query: str, n_results: int = 5) -> List[str]:
"""Retrieve relevant memories via semantic search."""
results = self.collection.query(
query_texts=[query],
n_results=n_results
)
return results["documents"][0] if results["documents"] else []
def recall_with_context(self, query: str, n_results: int = 3) -> str:
"""Retrieve memories formatted as context for an agent prompt."""
memories = self.recall(query, n_results)
if not memories:
return "No relevant past memories found."
context = "Relevant past memories:\n"
for i, mem in enumerate(memories, 1):
context += f"{i}. {mem}\n"
return context
def forget_old_memories(self, days_threshold: int = 30):
"""Archive or delete memories older than threshold."""
# ChromaDB stores metadata, so we can filter by date
cutoff = (datetime.now() - datetime.timedelta(days=days_threshold)).isoformat()
# Implementation depends on how you store dates in metadata
print(f"Archiving memories older than {cutoff}")
def get_recent_context(self, limit: int = 10) -> str:
"""Get the most recent memories as context."""
results = self.collection.get(limit=limit, include=["documents"])
if results["documents"]:
return "Recent context:\n" + "\n".join(
f"- {doc}" for doc in results["documents"]
)
return "No recent context available."
# Usage with an agent
class MemoryAwareAgent(ToolEnabledAgent):
"""Agent that consults its memory before acting."""
def __init__(self, name: str, system_prompt: str, tool_registry: ToolRegistry):
super().__init__(system_prompt, tool_registry)
self.memory = AgentMemory(name)
self.name = name
def execute_with_memory(self, user_request: str) -> str:
"""Execute a request, first consulting relevant memories."""
# Retrieve relevant past experiences
relevant_memories = self.memory.recall_with_context(user_request)
# Augment the system prompt with memory context
augmented_prompt = (
f"{self.system_prompt}\n\n"
f"{relevant_memories}\n\n"
f"Use the above memories to inform your response. "
f"After completing this task, store any important learnings."
)
# Temporarily override system prompt
original_prompt = self.system_prompt
self.system_prompt = augmented_prompt
result = self.execute_with_tools(user_request)
# Store the outcome as a new memory
self.memory.remember(
f"Request: {user_request}\nResponse: {result[:500]}\n"
f"Timestamp: {datetime.now().isoformat()}",
metadata={"type": "interaction", "timestamp": datetime.now().isoformat()}
)
self.system_prompt = original_prompt
return result
Multi-Agent Collaboration Patterns
Complex projects require multiple agents working together. Here are two proven collaboration patterns: the debate pattern for decision-making and the pipeline pattern for sequential workflows.
The Debate Pattern: Agents Critique Each Other
class DebateModerator:
"""Coordinates a debate between agents to reach better decisions."""
def __init__(self, agents: List[ToolEnabledAgent]):
self.agents = agents
def debate(self, question: str, rounds: int = 3) -> str:
"""Run a structured debate between agents."""
transcript = []
# Initial positions
print(f"šÆ Debate topic: {question}")
positions = []
for i, agent in enumerate(self.agents):
prompt = f"State your initial position and reasoning on this question:\n{question}"
position = agent.execute_with_tools(prompt)
positions.append(position)
transcript.append(f"Agent {i} (initial): {position}")
# Debate rounds
for round_num in range(rounds):
print(f"š Debate round {round_num + 1}/{rounds}")
for i, agent in enumerate(self.agents):
# Show this agent all other agents' latest positions
other_positions = [
f"Agent {j}: {positions[j]}"
for j in range(len(self.agents)) if j != i
]
critique_prompt = (
f"Here are the other agents' positions:\n"
f"{chr(10).join(other_positions)}\n\n"
f"Critique their arguments, identify weaknesses, "
f"and refine your own position on: {question}"
)
refined = agent.execute_with_tools(critique_prompt)
positions[i] = refined
transcript.append(f"Agent {i} (round {round_num + 1}): {refined}")
# Synthesis
synthesis_prompt = (
f"Debate transcript:\n{chr(10).join(transcript[-6:])}\n\n"
f"Synthesize the best arguments into a final, balanced recommendation "
f"on: {question}"
)
# Use the first agent as synthesizer
final = self.agents[0].execute_with_tools(synthesis_prompt)
return final
# Example: Code architecture decision
architect_agent = ToolEnabledAgent(
"You are a senior software architect favoring microservices.",
tool_registry
)
monolith_agent = ToolEnabledAgent(
"You are a pragmatic engineer favoring monolithic architectures for early-stage startups.",
tool_registry
)
moderator = DebateModerator([architect_agent, monolith_agent])
decision = moderator.debate(
"Should our SaaS product use microservices or a modular monolith? "
"We have 100 users, one developer, and need to move fast."
)
print("Final architecture recommendation:", decision)
The Pipeline Pattern: Sequential Processing
class AgentPipeline:
"""Processes tasks through a sequence of specialized agents."""
def __init__(self):
self.stages: List[tuple] = [] # (stage_name, agent, transform_fn)
def add_stage(self, name: str, agent: ToolEnabledAgent,
transform_fn: Callable = None):
"""Add a processing stage to the pipeline."""
self.stages.append((name, agent, transform_fn))
def process(self, initial_input: str) -> dict:
"""Run input through all pipeline stages sequentially."""
current_data = initial_input
stage_results = {}
for stage_name, agent, transform_fn in self.stages:
print(f"š¦ Pipeline stage: {stage_name}")
# Apply transformation to prepare input for this stage
if transform_fn:
stage_input = transform_fn(current_data)
else:
stage_input = current_data
# Execute this stage
result = agent.execute_with_tools(stage_input)
stage_results[stage_name] = result
current_data = result # Pass to next stage
return stage_results
# Example: Content publishing pipeline
research_agent = ToolEnabledAgent(
"You research topics thoroughly using web sources and compile structured findings.",
tool_registry
)
drafting_agent = ToolEnabledAgent(
"You write compelling, well-structured blog posts from research notes.",
tool_registry
)
editing_agent = ToolEnabledAgent(
"You edit content for clarity, grammar, SEO optimization, and readability.",
tool_registry
)
publishing_agent = ToolEnabledAgent(
"You format content for publication, create social media snippets, and schedule posts.",
tool_registry
)
pipeline = AgentPipeline()
pipeline.add_stage("research", research_agent)
pipeline.add_stage("drafting", drafting_agent)
pipeline.add_stage("editing", editing_agent)
pipeline.add_stage("publishing", publishing_agent)
results = pipeline.process(
"Create a blog post about AI-powered software development for solo founders. "
"Target audience: experienced developers considering going solo. "
"Length: 2000 words. Include practical examples."
)
for stage, output in results.items():
print(f"\n{'='*40}\nStage: {stage}\n{'='*40}\n{output[:300]}...")
Cost Management and Monitoring
When your workforce is AI-powered, your costs are directly tied to API calls. You need visibility into usage patterns and spending.
import time
from collections import defaultdict
class CostMonitor: