Multi-agent LLM orchestration patterns

planning_orchestrator.py

import random

# -------------------------------
# Simulated Agent Tools
# -------------------------------
class ResearchAgent:
    def run(self, topic, sources_required=3):
        print(f"[ResearchAgent] Researching: {topic} (need {sources_required} sources)")
        return f"Research data about {topic} from {sources_required} sources"

class VerifierAgent:
    def run(self, data, quality_level="high"):
        print(f"[VerifierAgent] Verifying with {quality_level} standards...")
        if "inaccurate" in data or (quality_level == "high" and random.random() < 0.4):
            return "FAILED"
        return "VERIFIED"

class SummarizerAgent:
    def run(self, data, word_limit=500):
        print(f"[SummarizerAgent] Summarizing to {word_limit} words...")
        if random.random() < 0.3:  # 30% failure rate
            return None
        return f"Summary ({word_limit} words): {data[:50]}..."

# -------------------------------
# Orchestrator Pattern
# -------------------------------
# NOTE: In real-world systems, the orchestrator would be an LLM-based agent
# with access to tools for dynamic planning/replanning. We show explicit 
# plan() and replan() methods here for educational illustration only.
# -------------------------------
class OrchestratorAgent:
    def __init__(self):
        self.researcher = ResearchAgent()
        self.verifier = VerifierAgent()
        self.summarizer = SummarizerAgent()
        self.max_attempts = 3

    def infer_goal(self, user_input):
        """Analyze input and determine what needs to be done"""
        return {"topic": user_input, "goal": "create_verified_summary"}

    def plan(self):
        """Define the sequence of agent actions"""
        return [
            ("research", self.researcher),
            ("verify", self.verifier), 
            ("summarize", self.summarizer)
        ]

    def goal_achieved(self, state):
        """Check if the main objective is complete"""
        return state.get("final_summary") is not None

    def replan(self, state, failed_step, attempt_number):
        """Adjust strategy when steps fail or goal isn't achieved"""
        print(f"[Orchestrator] Replanning after {failed_step} failure...")
        
        if failed_step == "verify":
            # Research again with different approach
            print("  → Strategy: Re-research with more reliable sources")
            return "research_retry"
        
        elif failed_step == "summarize":
            # Try different summarization approach
            print("  → Strategy: Use simpler summarization method")
            return "summarize_retry"
        
        elif attempt_number >= 2:
            # Change goal to something more achievable
            print("  → Strategy: Fallback to outline instead of full summary")
            return "fallback_outline"
        
        return "standard_retry"

    def execute(self, user_input, word_limit=500, sources_required=3, quality_level="high"):
        """
        Main orchestration loop with configurable parameters
        
        Args:
            user_input: The topic/request from user
            word_limit: Target length for final output
            sources_required: Minimum sources needed for research
            quality_level: "high", "medium", or "basic" - affects verification strictness
        """
        goal = self.infer_goal(user_input)
        plan = self.plan()
        state = {
            "topic": goal["topic"],
            "word_limit": word_limit,
            "sources_required": sources_required,
            "quality_level": quality_level
        }
        
        for attempt in range(self.max_attempts):
            print(f"\n--- Attempt {attempt + 1} ---")
            failed_step = None
            
            # Execute each planned step
            for step_name, agent in plan:
                if step_name == "research":
                    state["data"] = agent.run(state["topic"], sources_required)
                
                elif step_name == "verify":
                    verification = agent.run(state["data"], quality_level)
                    if verification == "FAILED":
                        failed_step = "verify"
                        break  # Exit inner loop to replan
                    state["verified_data"] = state["data"]
                
                elif step_name == "summarize":
                    summary = agent.run(state["verified_data"], word_limit)
                    if summary:
                        state["final_summary"] = summary
                    else:
                        failed_step = "summarize"
                        break  # Exit inner loop to replan
            
            # Check if goal is achieved
            if self.goal_achieved(state):
                print("✅ Goal achieved!")
                return state["final_summary"]
            
            # Replan based on what failed
            strategy = self.replan(state, failed_step, attempt)
            if strategy == "fallback_outline":
                break  # Exit to fallback
        
        # Fallback after max attempts
        print("⏰ Max attempts reached, using fallback")
        return f"Fallback outline: {state.get('verified_data', 'No verified data available')}"

# -------------------------------
# Demo: Automated Essay Writing
# -------------------------------
if __name__ == "__main__":
    orchestrator = OrchestratorAgent()
    
    # Example: Essay writing with specific requirements
    result = orchestrator.execute(
        user_input="youth unemployment",
        word_limit=800,           # Essay length requirement
        sources_required=5,       # Academic rigor requirement  
        quality_level="high"      # Publication-ready quality
    )
    
    print(f"\nFinal Essay Output: {result}")