Skip to content

Instantly share code, notes, and snippets.

@ndamulelonemakh
Last active June 19, 2025 11:13
Show Gist options
  • Save ndamulelonemakh/5eacd0145fd87d81511278d2d2442570 to your computer and use it in GitHub Desktop.
Save ndamulelonemakh/5eacd0145fd87d81511278d2d2442570 to your computer and use it in GitHub Desktop.
Multi-agent LLM orchestration patterns
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}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment