Examples
Production-ready examples showing real-world agent implementations
E-commerce Order Processing
intermediateMulti-agent system for processing orders, managing inventory, and handling shipments
Agents involved:
class OrderAgent(BaseAgent):
async def process_order(self, order):
# Validate order
if not await self.validate_order(order):
return {"status": "rejected"}
# Check inventory
await self.publish("inventory/check", {
"items": order["items"],
"order_id": order["id"]
})
# Wait for inventory response
response = await self.wait_for("inventory/response",
timeout=30)
if response["available"]:
await self.publish("orders/confirmed", order)
else:
await self.publish("orders/backorder", order)
Real-time Analytics Dashboard
advancedStream processing agents that aggregate and analyze data in real-time
Agents involved:
class AggregatorAgent(BaseAgent):
async def setup(self):
await self.subscribe("metrics/raw")
self.window = TimeWindow(seconds=60)
self.aggregations = {
'count': CountAggregator(),
'sum': SumAggregator(),
'avg': AvgAggregator(),
'p95': PercentileAggregator(95)
}
async def process_message(self, msg):
self.window.add(msg.data)
if self.window.is_complete():
results = {}
for name, agg in self.aggregations.items():
results[name] = agg.compute(self.window)
await self.publish("metrics/aggregated", {
"window": self.window.bounds(),
"metrics": results
})
Document Processing Pipeline
advancedIntelligent document processing with OCR, NLP, and data extraction
Agents involved:
class NLPAgent(BaseAgent):
async def setup(self):
self.model = await self.load_model("document-bert")
self.ner = await self.load_model("named-entity-recognition")
await self.subscribe("documents/text")
async def analyze_document(self, msg):
text = msg.data["text"]
doc_id = msg.data["doc_id"]
# Named entity recognition
entities = await self.ner.extract(text)
# Document classification
doc_type = await self.model.classify(text)
# Key information extraction
key_info = await self.extract_by_type(doc_type, text)
await self.publish("documents/analyzed", {
"doc_id": doc_id,
"type": doc_type,
"entities": entities,
"key_info": key_info
})
IoT Device Management
intermediateManage thousands of IoT devices with automatic scaling and monitoring
Agents involved:
class DeviceAgent(BaseAgent):
def __init__(self, device_id):
super().__init__(name=f"device-{device_id}")
self.device_id = device_id
self.metrics = MetricsCollector()
async def setup(self):
# Device-specific subscriptions
await self.subscribe(f"devices/{self.device_id}/commands")
await self.subscribe(f"devices/{self.device_id}/config")
await self.subscribe("devices/broadcast")
# Start monitoring
self.schedule_task(self.send_telemetry, interval=30)
self.schedule_task(self.check_health, interval=60)
async def send_telemetry(self):
data = await self.collect_metrics()
await self.publish(f"telemetry/{self.device_id}", {
"timestamp": datetime.now().isoformat(),
"metrics": data,
"status": self.get_status()
})
Collaborative Research Assistant
advancedMultiple AI agents working together to research topics and generate reports
Agents involved:
class ResearchAgent(BaseAgent):
async def research_topic(self, topic):
# Break down into subtopics
subtopics = await self.generate_subtopics(topic)
# Create research tasks
tasks = []
for subtopic in subtopics:
task_id = str(uuid.uuid4())
tasks.append({
"id": task_id,
"topic": subtopic,
"parent_topic": topic,
"depth": "detailed",
"sources_required": 5
})
await self.publish("research/tasks", {
"task": tasks[-1],
"deadline": datetime.now() + timedelta(hours=1)
})
# Coordinate results
await self.publish("research/coordinate", {
"topic": topic,
"task_ids": [t["id"] for t in tasks],
"output_format": "report"
})
Game Server Orchestration
intermediateManage game servers, matchmaking, and player statistics
Agents involved:
class MatchmakerAgent(BaseAgent):
def __init__(self):
super().__init__("matchmaker")
self.player_queue = PriorityQueue()
self.match_rules = MatchRules()
async def find_match(self, player):
# Add to queue with skill-based priority
priority = self.calculate_priority(player)
self.player_queue.add(player, priority)
# Try to form matches
if self.player_queue.size() >= self.match_size:
players = self.select_balanced_teams()
if players:
match = {
"id": str(uuid.uuid4()),
"players": players,
"mode": player["preferred_mode"],
"server": await self.allocate_server()
}
await self.publish("matches/created", match)
await self.notify_players(players, match)
Build Your Own
These examples are just the beginning. With ArtCafe.ai's flexible architecture, you can build any multi-agent system you can imagine.
Contribute Your Example
Have you built something amazing with ArtCafe.ai? We'd love to feature it here!
Learn how to contribute →