Workflow 002: Query Multiple Bots

Fresh 🌱

Workflow ID: WF-002
Version: 1.0
Status: Active

Overview

Query multiple Poe bots with the same question and compare their responses. Useful for benchmarking, getting diverse perspectives, or building comparison tools.

Prerequisites

• [x] API key obtained (SOP 001)
• [x] fastapi-poe installed (SOP 002)
• [x] Basic query understanding (SOP 003)

Workflow Steps

┌─────────────────────────────────────────────────────────────┐
│              MULTI-BOT QUERY WORKFLOW                         │
├─────────────────────────────────────────────────────────────┤
│                                                              │
│  1. Define Question ──► 2. Select Bots                      │
│        │                        │                            │
│        └──► 3. Query All ──────┘                            │
│        │                        │                            │
│        └──► 4. Compare Results ─┘                           │
│                                                              │
└─────────────────────────────────────────────────────────────┘

Procedure

Step 1: Define Your Question

question = "Explain quantum computing in simple terms"

Step 2: Select Bots to Query

bots = [
    "GPT-5",
    "Claude-Sonnet-4.5",
    "Grok-4"
]

Step 3: Query All Bots (Sequential)

import asyncio
import fastapi_poe as fp
import os
from typing import Dict

api_key = os.getenv("POE_API_KEY")

async def query_bot(bot_name: str, question: str) -> str:
    """Query a single bot."""
    message = fp.ProtocolMessage(role="user", content=question)
    
    response = ""
    async for partial in fp.get_bot_response(
        messages=[message],
        bot_name=bot_name,
        api_key=api_key
    ):
        response += partial
    
    return response

async def query_all_bots_sequential(bots: list, question: str) -> Dict[str, str]:
    """Query all bots one after another."""
    results = {}
    
    for bot in bots:
        print(f"Querying {bot}...")
        results[bot] = await query_bot(bot, question)
        print(f"✓ {bot} complete")
    
    return results

Step 4: Query All Bots (Parallel - Faster)

async def query_all_bots_parallel(bots: list, question: str) -> Dict[str, str]:
    """Query all bots simultaneously."""
    tasks = [query_bot(bot, question) for bot in bots]
    responses = await asyncio.gather(*tasks)
    
    return dict(zip(bots, responses))

Step 5: Compare Results

def compare_responses(results: Dict[str, str]):
    """Display and compare bot responses."""
    print("\n" + "="*80)
    print("COMPARISON RESULTS")
    print("="*80 + "\n")
    
    for bot, response in results.items():
        print(f"\n{'─'*80}")
        print(f"Bot: {bot}")
        print(f"{'─'*80}")
        print(response[:500] + "..." if len(response) > 500 else response)
        print(f"\nLength: {len(response)} characters")

Complete Example

import asyncio
import fastapi_poe as fp
import os
from typing import Dict, List
from datetime import datetime

api_key = os.getenv("POE_API_KEY")

class BotComparator:
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    async def query_bot(
        self,
        bot_name: str,
        question: str,
        parameters: dict = None
    ) -> Dict[str, any]:
        """Query a single bot and return metadata."""
        start_time = datetime.now()
        
        message = fp.ProtocolMessage(
            role="user",
            content=question,
            parameters=parameters or {}
        )
        
        response = ""
        async for partial in fp.get_bot_response(
            messages=[message],
            bot_name=bot_name,
            api_key=self.api_key
        ):
            response += partial
        
        end_time = datetime.now()
        duration = (end_time - start_time).total_seconds()
        
        return {
            "bot": bot_name,
            "response": response,
            "length": len(response),
            "duration": duration,
            "timestamp": end_time.isoformat()
        }
    
    async def compare_bots(
        self,
        bots: List[str],
        question: str,
        parallel: bool = True
    ) -> List[Dict[str, any]]:
        """Compare multiple bots."""
        print(f"Querying {len(bots)} bots: {', '.join(bots)}")
        print(f"Question: {question}\n")
        
        if parallel:
            print("Using parallel queries (faster)...")
            tasks = [
                self.query_bot(bot, question)
                for bot in bots
            ]
            results = await asyncio.gather(*tasks)
        else:
            print("Using sequential queries...")
            results = []
            for bot in bots:
                result = await self.query_bot(bot, question)
                results.append(result)
        
        return results
    
    def display_comparison(self, results: List[Dict[str, any]]):
        """Display formatted comparison."""
        print("\n" + "="*80)
        print("BOT COMPARISON RESULTS")
        print("="*80 + "\n")
        
        for result in results:
            print(f"\n{'─'*80}")
            print(f"Bot: {result['bot']}")
            print(f"Duration: {result['duration']:.2f}s")
            print(f"Response Length: {result['length']} characters")
            print(f"{'─'*80}")
            print(result['response'][:300] + "..." if len(result['response']) > 300 else result['response'])
        
        # Summary
        print("\n" + "="*80)
        print("SUMMARY")
        print("="*80)
        fastest = min(results, key=lambda x: x['duration'])
        longest = max(results, key=lambda x: x['length'])
        print(f"Fastest: {fastest['bot']} ({fastest['duration']:.2f}s)")
        print(f"Longest Response: {longest['bot']} ({longest['length']} chars)")

async def main():
    comparator = BotComparator(api_key)
    
    question = "What are the key differences between Python and JavaScript?"
    bots = ["GPT-5", "Claude-Sonnet-4.5", "Grok-4"]
    
    results = await comparator.compare_bots(bots, question, parallel=True)
    comparator.display_comparison(results)

if __name__ == "__main__":
    asyncio.run(main())

Advanced: With Custom Parameters

async def compare_with_parameters():
    comparator = BotComparator(api_key)
    
    question = "Explain machine learning"
    bots = [
        ("Claude-Sonnet-4.5", {"thinking_budget": 8192}),
        ("GPT-5", {"reasoning_effort": "high"})
    ]
    
    results = []
    for bot_name, params in bots:
        result = await comparator.query_bot(bot_name, question, params)
        results.append(result)
    
    comparator.display_comparison(results)

Verification

Expected output:

Querying 3 bots: GPT-5, Claude-Sonnet-4.5, Grok-4
Question: What is Python?

Using parallel queries (faster)...
Querying GPT-5...
Querying Claude-Sonnet-4.5...
Querying Grok-4...

================================================================================
BOT COMPARISON RESULTS
================================================================================
...

Best Practices

1. Use parallel queries for speed (unless rate limiting)
2. Handle errors for individual bots (one failure shouldn't stop all)
3. Store results for later analysis
4. Respect rate limits - don't query too many bots at once
5. Compare fairly - use same question and parameters

Troubleshooting

Problem: Some bots fail
Solution:

• Wrap each query in try/except
• Continue with successful queries
• Log failures for debugging

Problem: Rate limit exceeded
Solution:

• Reduce number of parallel queries
• Add delays between batches
• Implement rate limiting

Problem: Inconsistent results
Solution:

• Use same question for all bots
• Avoid time-sensitive questions
• Run multiple times and average

See Also

• SOP 003: Basic Query
• SOP 005: Custom Parameters
• Workflow 001: Setup External Application

---

Next: 003: Leaderboard Integration