DSPy + Microsoft Agent Framework Integration
A comprehensive guide to the integration patterns between DSPy and Microsoft Agent Framework in AgenticFleet. This skill documents how to leverage DSPy's structured reasoning capabilities with the Agent Framework's orchestration primitives.
Overview
AgenticFleet combines DSPy for intelligent prompt optimization and structured outputs with Microsoft Agent Framework for reliable multi-agent orchestration. This integration enables:
- Typed Signatures: Pydantic-validated DSPy outputs for type-safe orchestration
- DSPy-Enhanced Agents: ChatAgent wrappers with Chain of Thought, ReAct, and Program of Thought reasoning
- Routing Cache: TTL-based caching of routing decisions to reduce latency
- GEPA Optimization: Offline genetic prompt algorithm optimization
- Checkpoint Storage: Workflow resumption via agent-framework storage
- Agent Handoffs: Direct agent-to-agent transfers with context preservation
Architecture
┌─────────────────────────────────────────────────────────────────┐
│ AgenticFleet Integration │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ DSPyReasoner│────►│ AgentFactory│────►│ ChatAgent │ │
│ │ (Signatures)│ │ (YAML Config) │ (Enhanced) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ ┌──────▼───────────────────▼───────────────────▼──────┐ │
│ │ Microsoft Agent Framework │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │ │
│ │ │ Workflow │ │AgentThread│ │CheckpointStorage │ │ │
│ │ └──────────┘ └──────────┘ └──────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Typed Signatures with Pydantic
Signature Definition Pattern
All DSPy signatures in AgenticFleet use Pydantic models for structured outputs:
python
# src/agentic_fleet/dspy_modules/signatures.py
import dspy
from pydantic import BaseModel, Field
from typing import Literal
class TaskAnalysis(dspy.Signature):
"""Analyze a task with structured output."""
task: str = dspy.InputField(desc="The user's task description")
analysis: TaskAnalysisOutput = dspy.OutputField(
desc="Structured analysis of the task"
)
class TaskAnalysisOutput(BaseModel):
"""Pydantic model for typed signature output."""
complexity: Literal["low", "medium", "high"] = Field(
description="Estimated task complexity"
)
required_capabilities: list[str] = Field(
description="List of required capabilities"
)
estimated_steps: int = Field(ge=1, le=50)
preferred_tools: list[str] = Field(default_factory=list)
needs_web_search: bool = Field(description="Whether web search needed")
reasoning: str = Field(description="Reasoning behind analysis")
Using TypedPredictor
python
# src/agentic_fleet/dspy_modules/reasoner.py
from dspy import TypedPredictor
class DSPyReasoner(dspy.Module):
def __init__(self):
super().__init__()
self.analyzer = TypedPredictor(TaskAnalysis)
def analyze(self, task: str) -> TaskAnalysisOutput:
result = self.analyzer(task=task)
return result.analysis
Field Validators
Normalize inputs with Pydantic validators:
python
class RoutingDecisionOutput(BaseModel):
assigned_to: list[str] = Field(min_length=1)
execution_mode: Literal["delegated", "sequential", "parallel"]
@field_validator("assigned_to", mode="before")
@classmethod
def normalize_agents(cls, v: str | list[str]) -> list[str]:
if isinstance(v, str):
return [a.strip() for a in v.split(",") if a.strip()]
return v
@field_validator("execution_mode", mode="before")
@classmethod
def normalize_mode(cls, v: str) -> str:
mapping = {
"delegate": "delegated",
"single": "delegated",
"sequence": "sequential",
"concurrent": "parallel",
}
return mapping.get(v.strip().lower(), v)
DSPy Assertions for Validation
Hard and Soft Constraints
DSPy 3.x provides two assertion types for routing validation:
python
# src/agentic_fleet/dspy_modules/assertions.py
import dspy
# Hard constraint: causes backtracking on failure
dspy.Assert(condition, "error message")
# Soft constraint: guides optimization without failure
dspy.Suggest(condition, "guidance message")
Agent Assignment Validation
python
def validate_agent_exists(
assigned_agents: list[str],
available_agents: list[str]
) -> bool:
"""Check all assigned agents exist in available pool."""
# Hard constraint: must assign at least one agent
Assert(len(assigned_agents) > 0, "Must assign at least one agent")
# Soft suggestion: prefer matching case
for agent in assigned_agents:
Assert(
agent.lower() in [a.lower() for a in available_agents],
f"Agent '{agent}' not in available pool"
)
return True
Execution Mode Validation
python
def validate_execution_mode(
assigned_agents: list[str],
execution_mode: str
) -> bool:
"""Ensure execution mode matches agent count."""
if len(assigned_agents) > 1 and execution_mode == "delegated":
Suggest(
len(assigned_agents) == 1,
"Consider using 'parallel' for multiple agents"
)
return True
Usage in Signatures
python
class TaskRouting(dspy.Signature):
task: str = dspy.InputField(desc="The task to route")
team: str = dspy.InputField(desc="Available agents")
context: str = dspy.InputField(desc="Execution context")
decision: RoutingDecisionOutput = dspy.OutputField()
def __call__(self, task, team, context):
# Extract agent names from team description
available_agents = extract_agent_names(team)
# Validate before finalizing
result = super().__call__(task=task, team=team, context=context)
# Validate routing decision
validate_agent_exists(result.decision.assigned_to, available_agents)
validate_execution_mode(
result.decision.assigned_to,
result.decision.execution_mode
)
return result
DSPy-Enhanced Agents
Wrapping ChatAgent
python
# src/agentic_fleet/agents/base.py
from agent_framework._agents import ChatAgent
import dspy
class DSPyEnhancedAgent(ChatAgent):
def __init__(
self,
name: str,
chat_client,
instructions: str = "",
enable_dspy: bool = True,
reasoning_strategy: str = "chain_of_thought",
**kwargs
):
super().__init__(
name=name,
instructions=instructions,
chat_client=chat_client,
**kwargs
)
self.enable_dspy = enable_dspy
self.reasoning_strategy = reasoning_strategy
# Initialize reasoning modules
if enable_dspy:
self._init_reasoning_modules()
def _init_reasoning_modules(self):
"""Initialize DSPy reasoning strategies."""
if self.reasoning_strategy == "react":
self.react_module = dspy.ReAct(
"question -> answer",
tools=self.tools
)
elif self.reasoning_strategy == "program_of_thought":
self.pot_module = dspy.ProgramOfThought("question -> answer")
elif self.reasoning_strategy == "chain_of_thought":
self.cot_module = dspy.ChainOfThought("question -> answer")
Task Enhancement
python
class DSPyEnhancedAgent(ChatAgent):
def _enhance_task_with_dspy(self, task: str, context: str = "") -> str:
"""Enhance task using DSPy reasoning."""
if not self.enable_dspy:
return task
# Use Chain of Thought for complex tasks
enhancer = dspy.ChainOfThought("task, context -> enhanced_task")
result = enhancer(
task=task,
context=context or "No prior context"
)
return result.enhanced_task
async def run(self, message, **kwargs):
# Enhance task before execution
enhanced_message = self._enhance_task_with_dspy(
message,
kwargs.get("context", "")
)
# Run with enhanced task
return await super().run(enhanced_message, **kwargs)
Routing Cache
TTL-Based Cache Implementation
python
# src/agentic_fleet/dspy_modules/reasoner_cache.py
import time
from typing import Any
from collections import OrderedDict
class RoutingCache:
"""TTL-based cache for routing decisions."""
def __init__(self, ttl_seconds: int = 300, max_size: int = 1024):
self.ttl = ttl_seconds
self.max_size = max_size
self._cache: OrderedDict[str, tuple[Any, float]] = OrderedDict()
def get(self, key: str) -> Any | None:
"""Get cached value if not expired."""
if key not in self._cache:
return None
value, timestamp = self._cache[key]
# Check TTL
if time.time() - timestamp > self.ttl:
del self._cache[key]
return None
# Move to end (LRU)
self._cache.move_to_end(key)
return value
def set(self, key: str, value: Any) -> None:
"""Cache value with current timestamp."""
# Evict oldest if at capacity
if len(self._cache) >= self.max_size:
self._cache.popitem(last=False)
self._cache[key] = (value, time.time())
def clear(self) -> None:
"""Clear all cached entries."""
self._cache.clear()
Integration with DSPyReasoner
python
# src/agentic_fleet/dspy_modules/reasoner.py
class DSPyReasoner(dspy.Module):
def __init__(self, enable_routing_cache: bool = True, **kwargs):
super().__init__()
self.enable_routing_cache = enable_routing_cache
self._routing_cache = RoutingCache(
ttl_seconds=kwargs.get("cache_ttl_seconds", 300),
max_size=kwargs.get("cache_max_entries", 1024)
)
def _generate_cache_key(
self,
task: str,
team: str,
context: str
) -> str:
"""Generate cache key from routing inputs."""
import hashlib
content = f"{task}:{team}:{context}"
return hashlib.md5(content.encode()).hexdigest()
def route(self, task: str, team: str, context: str) -> RoutingDecisionOutput:
# Check cache first
if self.enable_routing_cache:
cache_key = self._generate_cache_key(task, team, context)
cached = self._routing_cache.get(cache_key)
if cached:
return cached
# Execute routing
result = self.router(task=task, team=team, context=context)
decision = result.decision
# Cache result
if self.enable_routing_cache:
self._routing_cache.set(cache_key, decision)
return decision
GEPA Optimization
Configuration
yaml
# src/agentic_fleet/config/workflow_config.yaml
dspy:
optimization:
enabled: true
examples_path: src/agentic_fleet/data/supervisor_examples.json
use_gepa: true
gepa_auto: light # light|medium|heavy
gepa_reflection_model: gpt-5-mini
gepa_history_min_quality: 8.0
gepa_history_limit: 200
gepa_val_split: 0.2
gepa_seed: 13
gepa_log_dir: .var/logs/dspy/gepa
Optimization Command
bash
# Run GEPA optimization
agentic-fleet optimize