ml-multi-model-orchestration

Coordinating multiple LLMs in pipelines, ensembles, and cascading workflows

Multi-Model Orchestration

Scope: Pipeline patterns, ensemble methods, specialist routing, cascading, context management, error handling Lines: ~400 Last Updated: 2025-10-26

When to Use This Skill

Activate this skill when:

Building workflows that combine strengths of multiple models
Implementing consensus mechanisms for critical decisions
Cascading from fast models to accurate models based on confidence
Designing specialist systems where different models handle different subtasks
Chaining models in sequential pipelines (planning -> execution -> review)
Managing context and state across multi-model calls
Handling failures with fallback models
Optimizing for both quality and cost through model mixing

Core Concepts

Orchestration Patterns

1. Pipeline (Sequential):

Models execute in sequence, each using previous output
Example: Planner -> Coder -> Reviewer -> Refiner
Benefit: Specialization per stage
Cost: Cumulative latency

2. Ensemble (Parallel):

Multiple models process same input, outputs combined
Example: GPT-4o + Claude + Gemini -> Vote or merge
Benefit: Higher accuracy, reduced bias
Cost: 3x+ API costs

3. Specialist Routing:

Route to model specialized for subtask
Example: Code->Claude, Math->DeepSeek R1, General->GPT-4o
Benefit: Optimal model per task
Cost: Routing overhead

4. Cascade (Conditional):

Start with cheap/fast model, escalate if needed
Example: GPT-4o mini -> (if unsure) -> GPT-4o
Benefit: Cost savings with quality backstop
Cost: Complexity in confidence detection

5. Hybrid (Combined):

Mix patterns: Pipeline + Ensemble, Cascade + Specialist
Example: Route by topic -> Ensemble -> Pipeline
Benefit: Maximum flexibility
Cost: Complex orchestration logic

Context Management

Challenges:

Passing large context between models (token costs)
Maintaining conversation state across calls
Avoiding context window overflow
Preserving semantic coherence

Solutions:

Prompt chaining (minimal context passing)
Shared memory/cache layer (Redis, database)
Summarization between stages (lossy but efficient)
Context compression (extract key information)

Error Handling Strategies

Fallback Chains:

Primary model -> (on error) -> Fallback 1 -> Fallback 2 -> Default response

Retry with Backoff:

Attempt 1 (immediate) -> Attempt 2 (+1s) -> Attempt 3 (+3s) -> Fail gracefully

Graceful Degradation:

Try ensemble -> (on partial failure) -> Use successful responses -> (on total failure) -> Single model

Arize Phoenix Multi-Model Tracing

Capabilities:

Span-level tracing across model calls
Routing decision visibility
Cost/latency breakdown per model
Error rate tracking per model
A/B test comparison

Span Structure:

Root Span (orchestration)
├─ Span: routing_decision
├─ Span: model_call_1 (claude-3.5-sonnet)
│  ├─ Attribute: tokens_used
│  ├─ Attribute: latency_ms
│  └─ Attribute: cost_usd
├─ Span: model_call_2 (gpt-4o)
└─ Span: result_aggregation

Implementation Patterns

Pipeline Pattern: Plan-Code-Review

import openai
from typing import Dict, Any

class CodeGenerationPipeline:
    """Multi-stage pipeline: planning -> coding -> review"""

    def __init__(self):
        self.planner_model = "gpt-4o"
        self.coder_model = "claude-3-5-sonnet-20241022"
        self.reviewer_model = "gpt-4o"

    def plan(self, requirement: str) -> str:
        """Stage 1: Planning with GPT-4o"""
        response = openai.ChatCompletion.create(
            model=self.planner_model,
            messages=[{
                "role": "system",
                "content": "You are a software architect. Create a detailed implementation plan."
            }, {
                "role": "user",
                "content": f"Plan implementation for: {requirement}"
            }],
            temperature=0.7
        )
        return response.choices[0].message.content

    def code(self, plan: str) -> str:
        """Stage 2: Coding with Claude (best at code)"""
        response = openai.ChatCompletion.create(
            model=self.coder_model,
            messages=[{
                "role": "system",
                "content": "You are an expert programmer. Implement the plan precisely."
            }, {
                "role": "user",
                "content": f"Implement this plan:\n\n{plan}"
            }],
            temperature=0.2
        )
        return response.choices[0].message.content

    def review(self, code: str, plan: str) -> Dict[str, Any]:
        """Stage 3: Review with GPT-4o"""
        response = openai.ChatCompletion.create(
            model=self.reviewer_model,
            messages=[{
                "role": "system",
                "content": "You are a code reviewer. Assess quality, correctness, and adherence to plan."
            }, {
                "role": "user",
                "content": f"Review this code against the plan.\n\nPlan:\n{plan}\n\nCode:\n{code}"
            }],
            temperature=0.3
        )

        review_text = response.choices[0].message.content

        # Extract approval (simple heuristic)
        approved = "approved" in review_text.lower() or "looks good" in review_text.lower()

        return {
            "approved": approved,
            "review": review_text,
            "code": code
        }

    def execute(self, requirement: str) -> Dict[str, Any]:
        """Execute full pipeline"""
        print("Stage 1: Planning...")
        plan = self.plan(requirement)

        print("Stage 2: Coding...")
        code = self.code(plan)

        print("Stage 3: Review...")
        result = self.review(code, plan)

        return {
            "requirement": requirement,
            "plan": plan,
            "code": result["code"],
            "review": result["review"],
            "approved": result["approved"]
        }

# Usage
pipeline = CodeGenerationPipeline()
result = pipeline.execute("Build a REST API for user authentication with JWT")

print(f"Approved: {result['approved']}")
print(f"Code:\n{result['code'][:500]}...")

Ensemble Pattern: Consensus Voting

import asyncio
from typing import List, Dict
from collections import Counter

class EnsembleOrchestrator:
    """Run multiple models in parallel and vote on results"""

    def __init__(self, models: List[str] = None):
        self.models = models or [
            "gpt-4o",
            "claude-3-5-sonnet-20241022",
            "gemini-2.5-pro"
        ]

    async def call_model(self, model: str, prompt: str) -> str:
        """Call single model asynchronously"""
        response = await asyncio.to_thread(
            openai.ChatCompletion.create,
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )
        return response.choices[0].message.content

    async def ensemble_vote(
        self,
        prompt: str,
        extract_answer_fn = None
    ) -> Dict[str, Any]:
        """Run ensemble and vote on answers"""

        # Call all models in parallel
        tasks = [
            self.call_model(model, prompt)
            for model in self.models
        ]
        responses = await asyncio.gather(*tasks)

        # Extract answers (default: use full response)
        if extract_answer_fn is None:
            extract_answer_fn = lambda x: x.strip().lower()

        answers = [extract_answer_fn(r) for r in responses]

        # Vote
        vote_counts = Counter(answers)
        majority_answer = vote_counts.most_common(1)[0][0]
        majority_count = vote_counts[majority_answer]

        # Check consensus (>50%)
        consensus = majority_count > len(self.models) / 2

        return {
            "models": self.models,
            "responses": responses,
            "answers": answers,
            "majority_answer": majority_answer,
            "vote_counts": dict(vote_counts),
            "consensus": consensus,
            "confidence": majority_count / len(self.models)
        }

# Usage
async def main():
    ensemble = EnsembleOrchestrator()

    # Multiple choice question
    prompt = """
    What is the capital of Australia?
    A) Sydney
    B) Melbourne
    C) Canberra
    D) Brisbane

    Answer with just the letter (A, B, C, or D).
    """

    def extract_letter(response: str) -> str:
        # Extract first letter A-D
        import re
        match = re.search(r'[ABCD]', response.upper())
        return match.group(0) if match else ""

    result = await ensemble.ensemble_vote(prompt, extract_answer_fn=extract_letter)

    print(f"Majority answer: {result['majority_answer']}")
    print(f"Consensus: {result['consensus']} ({result['confidence']:.0%})")
    print(f"Vote counts: {result['vote_counts']}")

asyncio.run(main())

Cascade Pattern: Confidence-Based Escalation

import re
from typing import Optional, Dict, Any

class CascadeOrchestrator:
    """Start with cheap model, escalate to expensive if uncertain"""

    def __init__(
        self,
        fast_model: str = "gpt-4o-mini",
        strong_model: str = "gpt-4o",
        confidence_threshold: float = 0.8
    ):
        self.fast = fast_model
        self.strong = strong_model
        self.threshold = confidence_threshold

    def extract_confidence(self, response: str) -> Optional[float]:
        """Extract confidence from response (if model provides it)"""
        # Look for patterns like "Confidence: 85%" or "(confidence: 0.85)"
        patterns = [
            r"confidence[:\s]+(\d+(?:\.\d+)?)\%?",
            r"\(confidence[:\s]+(\d+(?:\.\d+)?)\)",
        ]

        for pattern in patterns:
            match = re.search(pattern, response.lower())
            if match:
                conf = float(match.group(1))
                return conf / 100 if conf > 1 else conf

        return None

    def call_with_confidence(
        self,
        model: str,
        prompt: str,
        request_confidence: bool = True
    ) -> Dict[str, Any]:
        """Call model and extract confidence"""

        if request_confidence:
            prompt += "\n\nProvide your confidence level (0-100%) at the end."

        response = openai.ChatCompletion.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.3
        )

        text = response.choices[0].message.content
        confidence = self.extract_confidence(text)

        return {
            "model": model,
            "response": text,
            "confidence": confidence,
            "tokens": response.usage.total_tokens
        }

    def cascade(self, prompt: str) -> Dict[str, Any]:
        """Execute cascade: fast -> strong if needed"""

        print(f"Trying fast model: {self.fast}")
        fast_result = self.call_with_confidence(self.fast, prompt)

        # Check if confident enough
        if fast_result["confidence"] and fast_result["confidence"] >= self.threshold:
            print(f"Fast model confident ({fast_result['confidence']:.0%}), using result")
            return {
                **fast_result,
                "escalated": False,
                "total_tokens": fast_result["tokens"]
            }

        # Escalate to strong model
        print(f"Low confidence ({fast_result['confidence']:.0%}), escalating to {self.strong}")
        strong_result = self.call_with_confidence(self.strong, prompt)

        return {
            **strong_result,
            "escalated": True,
            "fast_result": fast_result,
            "total_tokens": fast_result["tokens"] + strong_result["tokens"]
        }

# Usage
cascade = CascadeOrchestrator(confidence_threshold=0.8)

result = cascade.cascade("What are the ethical implications of AGI development?")

print(f"Final model: {result['model']}")
print(f"Escalated: {result['escalated']}")
print(f"Total tokens: {result['total_tokens']}")

Specialist Routing with Context Passing

from typing import List, Dict, Any

class SpecialistOrchestrator:
    """Route subtasks to specialist models"""

    def __init__(self):
        self.specialists = {
            "code": "claude-3-5-sonnet-20241022",
            "math": "deepseek-r1-distilled-70b",
            "general": "gpt-4o",
            "fast": "gemini-2.5-flash"
        }
        self.conversation_history: List[Dict] = []

    def detect_task_type(self, query: str) -> str:
        """Simple task detection (use semantic routing for production)"""
        query_lower = query.lower()

        if any(kw in query_lower for kw in ["code", "function", "implement", "debug"]):
            return "code"
        if any(kw in query_lower for kw in ["calculate", "solve", "math", "equation"]):
            return "math"
        if len(query) < 50:
            return "fast"

        return "general"

    def call_specialist(
        self,
        task_type: str,
        query: str,
        include_history: bool = False
    ) -> Dict[str, Any]:
        """Route to specialist and maintain context"""

        model = self.specialists[task_type]

        # Build messages with optional history
        messages = []
        if include_history:
            messages.extend(self.conversation_history)

        messages.append({"role": "user", "content": query})

        response = openai.ChatCompletion.create(
            model=model,
            messages=messages,
            temperature=0.3
        )

        answer = response.choices[0].message.content

        # Update conversation history
        self.conversation_history.append({"role": "user", "content": query})
        self.conversation_history.append({"role": "assistant", "content": answer})

        return {
            "task_type": task_type,
            "model": model,
            "query": query,
            "answer": answer,
            "tokens": response.usage.total_tokens
        }

    def multi_turn_conversation(self, queries: List[str]) -> List[Dict[str, Any]]:
        """Handle multi-turn conversation with specialist routing"""

        results = []
        for query in queries:
            task_type = self.detect_task_type(query)
            print(f"Query: {query[:60]}... -> {task_type} ({self.specialists[task_type]})")

            result = self.call_specialist(
                task_type,
                query,
                include_history=True  # Maintain context
            )
            results.append(result)

        return results

# Usage
orchestrator = SpecialistOrchestrator()

conversation = [
    "Write a function to calculate Fibonacci numbers",
    "Now optimize it for large inputs",
    "What's the time complexity?",
    "Calculate the 50th Fibonacci number using your implementation"
]

results = orchestrator.multi_turn_conversation(conversation)

for r in results:
    print(f"\n[{r['task_type']}] {r['model']}")
    print(f"Answer: {r['answer'][:200]}...")

Phoenix Observability Integration

import phoenix as px
from phoenix.trace import using_project
from opentelemetry import trace

# Start Phoenix
px.launch_app()

tracer = trace.get_tracer(__name__)

@using_project("multi-model-orchestration")
def orchestrated_pipeline(requirement: str):
    """Pipeline with full observability"""

    with tracer.start_as_current_span("pipeline") as root_span:
        root_span.set_attribute("requirement", requirement)

        # Stage 1: Planning
        with tracer.start_as_current_span("stage_planning") as plan_span:
            plan_span.set_attribute("model", "gpt-4o")

            plan_response = openai.ChatCompletion.create(
                model="gpt-4o",
                messages=[{
                    "role": "user",
                    "content": f"Plan: {requirement}"
                }]
            )

            plan = plan_response.choices[0].message.content
            plan_span.set_attribute("tokens", plan_response.usage.total_tokens)
            plan_span.set_attribute("cost_usd", calculate_cost(plan_response, "gpt-4o"))

        # Stage 2: Coding
        with tracer.start_as_current_span("stage_coding") as code_span:
            code_span.set_attribute("model", "claude-3-5-sonnet")

            code_response = openai.ChatCompletion.create(
                model="claude-3-5-sonnet-20241022",
                messages=[{
                    "role": "user",
                    "content": f"Implement:\n{plan}"
                }]
            )

            code = code_response.choices[0].message.content
            code_span.set_attribute("tokens", code_response.usage.total_tokens)
            code_span.set_attribute("cost_usd", calculate_cost(code_response, "claude-3-5-sonnet"))

        # Stage 3: Review
        with tracer.start_as_current_span("stage_review") as review_span:
            review_span.set_attribute("model", "gpt-4o")

            review_response = openai.ChatCompletion.create(
                model="gpt-4o",
                messages=[{
                    "role": "user",
                    "content": f"Review:\nPlan: {plan}\nCode: {code}"
                }]
            )

            review = review_response.choices[0].message.content
            review_span.set_attribute("tokens", review_response.usage.total_tokens)
            review_span.set_attribute("cost_usd", calculate_cost(review_response, "gpt-4o"))

        # Calculate total metrics
        total_cost = (
            calculate_cost(plan_response, "gpt-4o") +
            calculate_cost(code_response, "claude-3-5-sonnet") +
            calculate_cost(review_response, "gpt-4o")
        )
        total_tokens = (
            plan_response.usage.total_tokens +
            code_response.usage.total_tokens +
            review_response.usage.total_tokens
        )

        root_span.set_attribute("total_cost_usd", total_cost)
        root_span.set_attribute("total_tokens", total_tokens)

        return {
            "plan": plan,
            "code": code,
            "review": review,
            "cost": total_cost,
            "tokens": total_tokens
        }

def calculate_cost(response, model: str) -> float:
    """Calculate cost based on token usage"""
    pricing = {
        "gpt-4o": {"input": 5.0, "output": 20.0},
        "claude-3-5-sonnet": {"input": 3.0, "output": 15.0},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "gemini-2.5-flash": {"input": 0.10, "output": 0.40}
    }

    model_pricing = pricing.get(model, pricing["gpt-4o"])
    usage = response.usage

    cost = (
        (usage.prompt_tokens / 1_000_000) * model_pricing["input"] +
        (usage.completion_tokens / 1_000_000) * model_pricing["output"]
    )

    return cost

Quick Reference

Orchestration Pattern Selection

Use Case                          | Recommended Pattern
----------------------------------|------------------------------------
Sequential specialization         | Pipeline
Critical decision (safety)        | Ensemble with voting
Cost optimization                 | Cascade (cheap -> expensive)
Domain-specific routing           | Specialist routing
Complex multi-step workflows      | Hybrid (Pipeline + Specialist)
High-reliability requirements     | Ensemble + Fallback chains

Context Passing Strategies

Strategy           | Cost   | Latency | Use Case
-------------------|--------|---------|---------------------------
Full history       | High   | High    | Short conversations (<10 turns)
Summarization      | Medium | Medium  | Long conversations
Key extraction     | Low    | Low     | Factual pipelines
Shared state (DB)  | Low    | Medium  | Multi-user systems
No context         | Lowest | Lowest  | Independent tasks

Error Handling Patterns

# Pattern 1: Fallback chain
try:
    response = call_model("claude-3-5-sonnet", prompt)
except Exception:
    try:
        response = call_model("gpt-4o", prompt)
    except Exception:
        response = call_model("gpt-4o-mini", prompt)

# Pattern 2: Retry with exponential backoff
for attempt in range(3):
    try:
        response = call_model("gpt-4o", prompt)
        break
    except Exception as e:
        if attempt == 2:
            raise
        time.sleep(2 ** attempt)

# Pattern 3: Graceful degradation
results = []
for model in ["gpt-4o", "claude-3-5-sonnet", "gemini-2.5-pro"]:
    try:
        results.append(call_model(model, prompt))
    except Exception:
        pass  # Continue with other models

if results:
    return aggregate(results)
else:
    return default_response()

Anti-Patterns

❌ Passing full context between all stages: Wastes tokens and money ✅ Summarize or extract key information between pipeline stages

❌ Ensemble without voting mechanism: Can't resolve disagreements ✅ Implement majority voting or weighted consensus

❌ Cascade without confidence scores: Can't decide when to escalate ✅ Prompt models for confidence or use logprobs for uncertainty estimation

❌ No fallback on model failures: Single point of failure ✅ Implement fallback chains with multiple model options

❌ Ignoring cost in orchestration: Ensemble costs 3x per query ✅ Track costs per pattern; use ensembles only for critical decisions

❌ Synchronous calls in ensemble: 3x latency penalty ✅ Use async/parallel calls for ensemble and specialist routing

❌ No observability in multi-model flows: Can't debug or optimize ✅ Use Arize Phoenix or similar for span-level tracing

❌ Hard-coded model names in orchestration: Brittle to model updates ✅ Use configuration or model registry for flexibility

Hanzo SDK: Simplified Multi-Model Orchestration

Instead of manually implementing orchestration patterns, use Hanzo SDK for automatic multi-model coordination with built-in routing, caching, and failover.

Pipeline Pattern with Hanzo

from hanzo import Hanzo

# Initialize with different strategies for different stages
hanzo_fast = Hanzo(routing_strategy='latency_optimized')
hanzo_smart = Hanzo(routing_strategy='quality_focused')

async def research_pipeline(topic: str) -> str:
    # Stage 1: Fast outline generation
    outline = await hanzo_fast.chat.completions.create_async(
        messages=[{
            'role': 'user',
            'content': f'Create a research outline for: {topic}'
        }]
    )
    
    # Stage 2: Quality content generation
    content = await hanzo_smart.chat.completions.create_async(
        messages=[{
            'role': 'user',
            'content': f'Write detailed content for: {outline.content}'
        }]
    )
    
    return content.choices[0].message.content

# Hanzo automatically:
# - Routed stage 1 to fastest model (Gemini Flash)
# - Routed stage 2 to best model (GPT-4)
# - Handled retries and fallbacks
# - Tracked costs per stage

Ensemble Pattern with Hanzo

from hanzo import Hanzo

async def ensemble_decision(query: str) -> str:
    # Hanzo can query multiple models in parallel
    hanzo = Hanzo(inference_mode='hybrid')
    
    # Parallel calls to different models
    responses = await hanzo.chat.completions.batch_create_async([
        {'messages': [{'role': 'user', 'content': query}], 'model': 'gpt-4'},
        {'messages': [{'role': 'user', 'content': query}], 'model': 'claude-3-5-sonnet'},
        {'messages': [{'role': 'user', 'content': query}], 'model': 'gemini-2.5-pro'}
    ])
    
    # Extract answers
    answers = [r.choices[0].message.content for r in responses]
    
    # Aggregate with another model
    final = await hanzo.chat.completions.create_async(
        messages=[{
            'role': 'user',
            'content': f'Synthesize these expert opinions:\n\n' + '\n\n'.join(answers)
        }],
        model='gpt-4'  # Use best model for synthesis
    )
    
    return final.choices[0].message.content

Specialist Routing with Hanzo MCP

from hanzo import Hanzo
from hanzo.mcp import MCPClient

# Hanzo can use MCP tools for specialized tasks
hanzo = Hanzo(
    inference_mode='hybrid',
    mcp_servers=['http://localhost:8081']
)

mcp = MCPClient('http://localhost:8081')

async def route_by_intent(query: str) -> str:
    # First, classify intent
    intent = await hanzo.chat.completions.create_async(
        messages=[{
            'role': 'user',
            'content': f'Classify this query intent (code/search/analyze): {query}'
        }],
        model='gpt-4o-mini'  # Fast classifier
    )
    
    if 'code' in intent.content.lower():
        # Use code specialist model
        return await hanzo.chat.completions.create_async(
            messages=[{'role': 'user', 'content': query}],
            model='claude-3-5-sonnet'  # Best for code
        )
    elif 'search' in intent.content.lower():
        # Use MCP search tool
        tools = mcp.get_tools()
        return await hanzo.chat.completions.create_async(
            messages=[{'role': 'user', 'content': query}],
            tools=tools,
            tool_choice='auto'
        )
    else:
        # General analysis
        return await hanzo.chat.completions.create_async(
            messages=[{'role': 'user', 'content': query}],
            model='gpt-4'  # Best for analysis
        )

Cascade Pattern with Hanzo

from hanzo import Hanzo

hanzo_local = Hanzo(inference_mode='local')
hanzo_cloud = Hanzo(inference_mode='cloud')

async def cascade_query(query: str, complexity_threshold: float = 0.7) -> str:
    # Try fast local model first
    try:
        response = await hanzo_local.chat.completions.create_async(
            messages=[{'role': 'user', 'content': query}],
            model='llama-3-8b',
            timeout=5  # Fast timeout
        )
        
        # Check confidence (if model provides it)
        if hasattr(response, 'confidence') and response.confidence > complexity_threshold:
            return response.choices[0].message.content
    except TimeoutError:
        pass  # Fall through to cloud
    
    # Escalate to cloud model
    response = await hanzo_cloud.chat.completions.create_async(
        messages=[{'role': 'user', 'content': query}],
        model='gpt-4'  # Best model for complex queries
    )
    
    return response.choices[0].message.content

Comparison: Manual Orchestration vs Hanzo SDK

| Pattern | Manual Implementation | Hanzo SDK | |---------|----------------------|-----------| | Pipeline | 50+ lines, manual routing | 10-15 lines, auto-routing | | Ensemble | Custom parallel execution | Built-in batch API | | Specialist | Manual intent classification | Automatic or MCP-based | | Cascade | Custom timeout/retry logic | Built-in with fallbacks | | Cost Tracking | Custom per-pattern | Automatic all patterns | | Caching | Manual between stages | Automatic semantic cache | | Observability | Manual OpenTelemetry | Built-in metrics | | Error Handling | Custom per model | Automatic retry/circuit breaker |

Advanced: Multi-Model Workflow

from hanzo import Hanzo
from hanzo.workflows import Pipeline

# Define reusable workflow
class ResearchWorkflow(Pipeline):
    def __init__(self):
        super().__init__()
        
        # Different Hanzo instances for different stages
        self.fast_hanzo = Hanzo(routing_strategy='latency_optimized')
        self.smart_hanzo = Hanzo(routing_strategy='quality_focused')
        self.local_hanzo = Hanzo(inference_mode='local')
    
    async def research(self, topic: str) -> dict:
        # Stage 1: Generate outline (fast)
        outline = await self.fast_hanzo.chat.completions.create_async(
            messages=[{'role': 'user', 'content': f'Outline: {topic}'}]
        )
        
        # Stage 2: Parallel research on each section
        sections = outline.content.split('\n')
        research_tasks = [
            self.local_hanzo.chat.completions.create_async(
                messages=[{'role': 'user', 'content': f'Research: {section}'}]
            )
            for section in sections
        ]
        research_results = await asyncio.gather(*research_tasks)
        
        # Stage 3: Synthesize with best model
        synthesis = await self.smart_hanzo.chat.completions.create_async(
            messages=[{
                'role': 'user',
                'content': f'Synthesize:\n\n' + '\n\n'.join([r.content for r in research_results])
            }]
        )
        
        return {
            'outline': outline.content,
            'research': [r.content for r in research_results],
            'synthesis': synthesis.content,
            'total_cost': sum([
                outline.cost,
                sum(r.cost for r in research_results),
                synthesis.cost
            ])
        }

# Use workflow
workflow = ResearchWorkflow()
result = await workflow.research('Quantum Computing Applications')
print(f"Total cost: ${result['total_cost']:.4f}")

Key Benefits:

10x less code: Hanzo SDK handles routing, retries, caching automatically
Built-in observability: Automatic cost tracking and performance metrics
Privacy-first: Local models via Hanzo Node before cloud escalation
Production-ready: Circuit breakers, retries, fallbacks included
Type-safe: Full TypeScript support with Python compatibility

See Also:

python-sdk.md - Complete Hanzo SDK documentation
hanzo-mcp.md - MCP tool integration for workflows
hanzo-node.md - Local model inference
llm-model-routing.md - Automatic routing strategies

Related Skills

llm-model-routing.md - Routing strategies and RouteLLM framework
llm-model-selection.md - Choosing models for capabilities and cost
api-error-handling.md - Robust error handling across providers
observability-distributed-tracing.md - OpenTelemetry and Phoenix
modal-web-endpoints.md - Deploying orchestration as FastAPI services
redis-caching.md - Shared state and caching for multi-model systems
python-sdk.md - Hanzo SDK for multi-model orchestration
hanzo-mcp.md - MCP tools for specialized tasks
hanzo-node.md - Local inference infrastructure

Last Updated: 2025-10-28 Format Version: 1.0 (Atomic)