Skip to content

Strategies

Strategies in the digitalNXT Agency framework encapsulate the business logic for coordinating multiple AI agents. They define how agents work together to solve complex problems, from simple sequential execution to sophisticated routing and parallel processing.

Core Concepts

📋 Strategy Pattern

Strategies implement the Strategy design pattern, allowing the Agency to: - Switch between different execution approaches at runtime - Maintain consistent interfaces while varying behavior - Extend functionality without modifying core framework code

🔄 Execution Lifecycle

Every strategy follows a standardized lifecycle: 1. Initialization: Setup agents and configuration 2. Execution: Run the strategy logic with timing and path tracking 3. Result Processing: Package outputs with metadata and performance data

Strategy Architecture

graph TD
    BaseStrategy[📝 Base Strategy] --> |implements| ConcreteStrategy[🎯 Concrete Strategy]

    ConcreteStrategy --> Agent1[🤖 Agent 1]
    ConcreteStrategy --> Agent2[🤖 Agent 2]
    ConcreteStrategy --> AgentN[🤖 Agent N]

    ConcreteStrategy --> |uses| Processor[⚙️ Processor]

    subgraph "Strategy Execution"
        ExecutionPath[📍 Execution Path]
        ExecutionTimer[⏱️ Execution Timer]
        ExtraData[📊 Extra Data]
    end

    ConcreteStrategy --> ExecutionPath
    ConcreteStrategy --> ExecutionTimer
    ConcreteStrategy --> ExtraData

    ConcreteStrategy --> |returns| StrategyResult[📊 Strategy Result]

    subgraph "Built-in Strategies"
        RouterStrategy[🚦 Router Strategy]
        SequentialStrategy[📅 Sequential Strategy]
        ParallelStrategy[⚡ Parallel Strategy]
    end

    BaseStrategy --> RouterStrategy
    BaseStrategy --> SequentialStrategy
    BaseStrategy --> ParallelStrategy

    classDef baseClass fill:#e1f5fe
    classDef strategyClass fill:#e8f5e8
    classDef agentClass fill:#fff3e0
    classDef executionClass fill:#fce4ec
    classDef builtinClass fill:#f3e5f5

    class BaseStrategy baseClass
    class ConcreteStrategy strategyClass
    class Agent1,Agent2,AgentN,Processor agentClass
    class ExecutionPath,ExecutionTimer,ExtraData,StrategyResult executionClass
    class RouterStrategy,SequentialStrategy,ParallelStrategy builtinClass

Base Strategy Class

🏗️ Abstract Structure

All strategies inherit from the Strategy base class:

from agency.strategies.base import Strategy
from agency.strategies.models import StrategyResult
from typing import Any

class CustomStrategy(Strategy):
    def _setup_agents(self) -> None:
        """Setup default agents with optional overrides."""
        self.primary_agent = self.overrides.get('primary_agent') or Agent(...)
        self.secondary_agent = self.overrides.get('secondary_agent') or Agent(...)

    async def _execute_strategy(self, input_data: str, context: dict[str, Any] = None) -> str:
        """Execute the core strategy logic."""
        self._add_to_execution_path("primary_processing")

        # Use processor for observability if available
        if self.processor:
            result = await self.processor.run(
                agent=self.primary_agent,
                input_data=input_data,
                context=context
            )
        else:
            # Fallback execution
            result = await Runner.run(self.primary_agent, input_data, context=context)

        return result.final_output

🔧 Core Features

Automatic Timing and Metadata

# The base Strategy class automatically provides:
result = await strategy.execute(input_data, context)

print(f"Execution time: {result.execution_time}s")
print(f"Path taken: {result.execution_path}")
print(f"Strategy type: {result.metadata['strategy_type']}")
print(f"Input/output lengths: {result.metadata['input_length']}/{result.metadata['output_length']}")

Agent Override System

# Strategies support runtime agent customization
custom_agent = Agent(name="Specialized Agent", instructions="...")

strategy = RouterStrategy(
    billing_agent=custom_agent,  # Override default billing agent
    # Other agents use defaults
)

Processor Integration

# Strategies automatically receive processor from Agency
class MyStrategy(Strategy):
    def _setup_agents(self):
        self.agent = Agent(name="My Agent")

    async def _execute_strategy(self, input_data: str, context: dict = None) -> str:
        # self.processor is automatically available
        # self.logger is automatically available from processor

        self.logger.info("Starting strategy execution")

        if self.processor:
            result = await self.processor.run(self.agent, input_data, context)
        else:
            self.logger.warning("No processor available, using direct runner")
            result = await Runner.run(self.agent, input_data, context=context)

        return result.final_output

💾 Memory and Session Integration

Strategies automatically receive session context when used with memory-enabled agencies:

class MemoryAwareStrategy(Strategy):
    """Strategy that leverages conversation memory."""

    def _setup_agents(self):
        self.agent = Agent(
            name="Memory-Aware Assistant",
            instructions="You are a helpful assistant with access to conversation history."
        )

    async def _execute_strategy(self, input_data: str, context: dict = None) -> str:
        """Execute with full session context."""
        self._add_to_execution_path("memory_aware_processing")

        # Session context is automatically available
        session_id = context.get('session_id') if context else None
        user_id = context.get('user_id') if context else None
        conversation_id = context.get('conversation_id') if context else None

        # Log session information
        if session_id:
            self.logger.info(f"Processing with session: {session_id}")

        # Enhanced context with memory information
        enhanced_context = {
            **(context or {}),
            "has_memory": session_id is not None,
            "conversation_context": conversation_id is not None
        }

        # Process with memory-aware context
        if self.processor:
            result = await self.processor.run(
                agent=self.agent,
                input_data=input_data,
                context=enhanced_context
            )
        else:
            result = await Runner.run(
                self.agent,
                input_data,
                context=enhanced_context
            )

        # Store memory-related metadata
        self.extra_data = {
            "session_enabled": session_id is not None,
            "user_context": user_id is not None,
            "conversation_tracking": conversation_id is not None
        }

        return result.final_output

# Usage with memory-enabled agency
from agency.factories.agency_factory import AgencyFactory
from agency.sessions.session_factory import SessionFactory

session = SessionFactory.create_file_session("user_123", "data/user_123.db")
agency = AgencyFactory.create_agency(
    config=AgencyConfig(name="MemorySupport"),
    session=session,
    user_id="user_123"
)

# Context automatically includes session information
result = await agency.run("Remember what we discussed about my project?")

Session Context Variables: - session_id: SQLiteSession identifier for conversation persistence - user_id: User identifier for tracking across sessions - conversation_id: High-level conversation identifier - message_count: Number of messages in current conversation - has_memory: Boolean indicating if memory is enabled

Built-in Strategies

🚦 Router Strategy

Routes requests to appropriate specialist agents based on content analysis.

graph LR
    Input[📥 User Input] --> Triage[🤖 Triage Agent]

    Triage --> |"billing query"| Billing[💰 Billing Agent]
    Triage --> |"technical issue"| Technical[🔧 Technical Agent]

    Billing --> Output1[📤 Billing Response]
    Technical --> Output2[📤 Technical Response]

    classDef inputClass fill:#e3f2fd
    classDef triageClass fill:#fff3e0
    classDef specialistClass fill:#e8f5e8
    classDef outputClass fill:#f3e5f5

    class Input inputClass
    class Triage triageClass
    class Billing,Technical specialistClass
    class Output1,Output2 outputClass

Example Usage: 

from agency.strategies.router import RouterStrategy

# Use default agents
strategy = RouterStrategy()

# Customize specific agents
strategy = RouterStrategy(
    billing_agent=CustomBillingAgent(),
    technical_agent=CustomTechnicalAgent()
)

# Get routing information
routing_info = strategy.get_routing_info()
print(routing_info)
# Output: {"triage": "Router Triage", "billing": "Billing Specialist", "technical": "Technical Specialist"}

Key Features: - Automatic Routing: Triage agent determines the appropriate specialist - Handoff System: Uses OpenAI Agent handoff functionality - Final Agent Tracking: Identifies which agent provided the final response - Specialist Agents: Pre-configured billing and technical specialists

📅 Sequential Strategy (Planned)

Executes agents in a predefined order, passing outputs as inputs to subsequent agents.

graph LR
    Input[📥 Input] --> Agent1[🤖 Agent 1]
    Agent1 --> Agent2[🤖 Agent 2]
    Agent2 --> Agent3[🤖 Agent 3]
    Agent3 --> Output[📤 Final Output]

    classDef inputClass fill:#e3f2fd
    classDef agentClass fill:#e8f5e8
    classDef outputClass fill:#f3e5f5

    class Input inputClass
    class Agent1,Agent2,Agent3 agentClass
    class Output outputClass

⚡ Parallel Strategy (Planned)

Executes multiple agents concurrently and combines their results.

graph TD
    Input[📥 Input] --> Agent1[🤖 Agent 1]
    Input --> Agent2[🤖 Agent 2]
    Input --> Agent3[🤖 Agent 3]

    Agent1 --> Combiner[🔄 Result Combiner]
    Agent2 --> Combiner
    Agent3 --> Combiner

    Combiner --> Output[📤 Combined Output]

    classDef inputClass fill:#e3f2fd
    classDef agentClass fill:#e8f5e8
    classDef combinerClass fill:#fff3e0
    classDef outputClass fill:#f3e5f5

    class Input inputClass
    class Agent1,Agent2,Agent3 agentClass
    class Combiner combinerClass
    class Output outputClass

Strategy Result Model

📊 Comprehensive Output

Every strategy execution returns a StrategyResult with rich metadata:

@dataclass
class StrategyResult:
    # Core result
    output: str                           # The main output from strategy

    # Execution metadata
    execution_time: float                 # Time taken in seconds
    execution_path: list[str]             # Steps that were executed

    # Optional metadata
    metadata: dict[str, Any]              # Strategy-specific metadata
    extra_data: dict[str, Any]            # Additional data and extensions
    created_at: datetime                  # When result was created

Example Result: 

result = StrategyResult(
    output="Your billing inquiry has been resolved. Current balance: $0.00",
    execution_time=2.34,
    execution_path=["triage_routing", "billing_agent_invoked"],
    metadata={
        "strategy_type": "RouterStrategy",
        "input_length": 45,
        "output_length": 89,
        "final_agent": "Billing Specialist"
    },
    extra_data={
        "processor": {"service": "Customer Support", "version": "1.0"},
        "performance": {"triage_time": 0.8, "billing_time": 1.5}
    }
)

Custom Strategy Development

🛠️ Creating Custom Strategies

Simple Custom Strategy

class GreetingStrategy(Strategy):
    def _setup_agents(self):
        """Setup greeting agent."""
        self.greeting_agent = self.overrides.get('greeting_agent') or Agent(
            name="Greeting Specialist",
            instructions="You are a friendly greeting agent. Provide warm, personalized greetings."
        )

    async def _execute_strategy(self, input_data: str, context: dict = None) -> str:
        """Execute greeting logic."""
        self._add_to_execution_path("greeting_generation")

        # Add context-aware processing
        user_name = context.get('user_name', 'there') if context else 'there'
        enhanced_input = f"Greet the user named {user_name}. Original message: {input_data}"

        if self.processor:
            result = await self.processor.run(
                agent=self.greeting_agent,
                input_data=enhanced_input,
                context=context
            )
        else:
            result = await Runner.run(self.greeting_agent, enhanced_input, context=context)

        # Store additional data
        self.extra_data = {"user_name": user_name, "greeting_type": "personalized"}

        return result.final_output

Advanced Custom Strategy with Multiple Agents

class ResearchStrategy(Strategy):
    def _setup_agents(self):
        """Setup research pipeline agents."""
        self.researcher = self.overrides.get('researcher') or Agent(
            name="Research Specialist",
            instructions="Research the given topic thoroughly and gather relevant information."
        )

        self.analyzer = self.overrides.get('analyzer') or Agent(
            name="Analysis Specialist",
            instructions="Analyze research findings and identify key insights."
        )

        self.synthesizer = self.overrides.get('synthesizer') or Agent(
            name="Synthesis Specialist",
            instructions="Synthesize analysis into a clear, actionable summary."
        )

    async def _execute_strategy(self, input_data: str, context: dict = None) -> str:
        """Execute multi-stage research process."""
        # Stage 1: Research
        self._add_to_execution_path("research_phase")
        research_result = await self._run_agent(self.researcher, input_data, context)

        # Stage 2: Analysis
        self._add_to_execution_path("analysis_phase")
        analysis_input = f"Analyze this research: {research_result}"
        analysis_result = await self._run_agent(self.analyzer, analysis_input, context)

        # Stage 3: Synthesis
        self._add_to_execution_path("synthesis_phase")
        synthesis_input = f"Synthesize this analysis into actionable insights: {analysis_result}"
        final_result = await self._run_agent(self.synthesizer, synthesis_input, context)

        # Store pipeline metadata
        self.extra_data = {
            "pipeline_stages": len(self.execution_path),
            "research_length": len(research_result),
            "analysis_length": len(analysis_result),
            "synthesis_length": len(final_result)
        }

        return final_result

    async def _run_agent(self, agent, input_data: str, context: dict) -> str:
        """Helper method to run agents with processor."""
        if self.processor:
            result = await self.processor.run(agent, input_data, context)
        else:
            result = await Runner.run(agent, input_data, context=context)
        return result.final_output

🔧 Strategy Best Practices

✅ Do

# Use descriptive execution path tracking
self._add_to_execution_path("user_authentication")
self._add_to_execution_path("data_retrieval")
self._add_to_execution_path("response_generation")

# Provide meaningful extra_data
self.extra_data = {
    "authentication_method": "oauth2",
    "data_source": "customer_database",
    "response_confidence": 0.95
}

# Handle processor availability gracefully
if self.processor:
    result = await self.processor.run(agent, input_data, context)
else:
    self.logger.warning("No processor available, using direct runner")
    result = await Runner.run(agent, input_data, context=context)

# Use override system for flexibility
self.main_agent = self.overrides.get('main_agent') or default_agent

❌ Avoid

# Don't hardcode agent configurations
self.agent = Agent(name="Hardcoded Agent")  # Not customizable

# Don't ignore execution path tracking
# Missing: self._add_to_execution_path(...)

# Don't assume processor availability
result = await self.processor.run(...)  # May be None

# Don't forget error handling
try:
    result = await self._execute_some_logic()
except Exception:
    pass  # Silent failures lose valuable debugging info

Strategy Integration

🔌 Using Strategies with Agency

# Basic integration
strategy = RouterStrategy()
agency = Agency(name="Support Agency", strategy=strategy)

# Advanced integration with overrides and context
custom_strategy = ResearchStrategy(
    researcher=CustomResearchAgent(),
    analyzer=AdvancedAnalysisAgent()
)

agency = Agency(
    name="Research Agency",
    strategy=custom_strategy,
    context={"domain": "technology", "depth": "comprehensive"},
    langfuse_mode=LangfuseMode.AUTO
)

# Execution with rich context
result = await agency.run(
    "Research the impact of AI on software development",
    run_context={"priority": "high", "deadline": "2024-01-30"}
)

🔄 Strategy Lifecycle Management

# Strategies are designed to be recreated for each major workflow change
# but can handle multiple runs with the same configuration

strategy = RouterStrategy()
agency = Agency(name="Support", strategy=strategy)

# Multiple runs with same strategy instance
result1 = await agency.run("Billing question")
result2 = await agency.run("Technical issue")
result3 = await agency.run("General inquiry")

# Each run maintains separate execution paths and timing
assert result1.execution_path != result2.execution_path
assert result1.execution_time != result2.execution_time

Performance Considerations

⚡ Optimization Tips

  • Agent Reuse: Agents are initialized once during strategy setup
  • Processor Caching: Processor connections are reused across runs
  • Context Efficiency: Avoid large context objects that are serialized frequently
  • Execution Path: Keep execution paths descriptive but concise

📊 Monitoring Strategy Performance

# Analyze strategy performance from results
results = []
for input_data in test_inputs:
    result = await agency.run(input_data)
    results.append({
        'execution_time': result.execution_time,
        'execution_path': result.execution_path,
        'input_length': len(input_data),
        'output_length': len(result.output)
    })

# Identify performance patterns
avg_time = sum(r['execution_time'] for r in results) / len(results)
common_paths = Counter(tuple(r['execution_path']) for r in results)

Strategy Selection Guide

Choose RouterStrategy when: - You need to route different types of requests to specialists - You want automatic request classification - You have distinct agent capabilities for different domains

Choose SequentialStrategy when: - You have a multi-step process that must be executed in order - Output from one step becomes input to the next - You need pipeline-style processing

Choose ParallelStrategy when: - Multiple agents can work on the same input simultaneously - You want to combine different perspectives or approaches - Speed is important and agents can run concurrently

Custom Strategy Development

When developing custom strategies, focus on: - Clear separation between setup (_setup_agents) and execution (_execute_strategy) - Meaningful execution path tracking for debugging - Proper error handling and graceful degradation - Rich metadata in extra_data for observability