Introduction
Intelligent agents are autonomous systems capable of perceiving their environment, reasoning about goals, and taking actions to achieve them. DSPy provides a structured way to implement the core loops of perception, planning, and execution that define agency.
Understanding Intelligent Agents
Core Components
- Perception: Understanding the current state (e.g., reading user input, checking database).
- Planning: Breaking down goals into actionable steps.
- Decision Making: Choosing the best action from available options.
- Memory: Retaining context from past interactions.
- Execution: Performing actions (e.g., API calls, database updates).
Agent Types
- Reactive: Responds immediately to inputs without deep planning.
- Proactive: Plans ahead to achieve long-term goals.
- Collaborative: Works with other agents or humans.
- Learning: Improves strategies based on feedback.
Building Agents with DSPy
Basic Reactive Agent
A simple agent that perceives, decides, and acts:
Python
import dspy
class ReactiveAgent(dspy.Module):
def __init__(self, name, capabilities):
super().__init__()
self.name = name
self.capabilities = capabilities
self.perceive = dspy.Predict("input -> perceived_state")
self.decide = dspy.Predict("state, capabilities -> action, reasoning")
self.memory = {}
def forward(self, input_text):
# Perception
perception = self.perceive(input=input_text)
current_state = perception.perceived_state
# Decision
decision = self.decide(
state=current_state,
capabilities=", ".join(self.capabilities)
)
return dspy.Prediction(
agent_name=self.name,
perceived_state=current_state,
action=decision.action,
reasoning=decision.reasoning
)Proactive Agent with Planning
An agent that creates and executes multi-step plans:
Python
class ProactiveAgent(dspy.Module):
def __init__(self, name, goals, tools):
super().__init__()
self.name = name
self.goals = goals
self.tools = tools
self.understand_context = dspy.Predict("input -> context, user_intent")
self.create_plan = dspy.ChainOfThought("context, intent, goals, tools -> plan")
self.execute_step = dspy.Predict("plan, current_step, tools -> action, next_step")
self.current_plan = None
self.current_step = 0
def forward(self, input_text):
# Understand context
understanding = self.understand_context(input=input_text)
# Create/Update Plan
if not self.current_plan or "new" in understanding.user_intent:
planning = self.create_plan(
context=understanding.context,
intent=understanding.user_intent,
goals=", ".join(self.goals),
tools=", ".join(self.tools)
)
self.current_plan = planning.plan
self.current_step = 0
# Execute
execution = self.execute_step(
plan=self.current_plan,
current_step=str(self.current_step),
tools=", ".join(self.tools)
)
if execution.next_step:
self.current_step = int(execution.next_step)
return dspy.Prediction(
action=execution.action,
plan=self.current_plan,
step=self.current_step
)Real-World Application: Customer Service Agent
A practical agent handling customer inquiries, knowledge base search, and escalation:
Python
class CustomerServiceAgent(dspy.Module):
def __init__(self, company_name, knowledge_base):
super().__init__()
self.company_name = company_name
self.knowledge_base = knowledge_base
self.classify_intent = dspy.Predict("customer_message -> intent, urgency, sentiment")
self.search_knowledge = dspy.Retrieve(k=3)
self.generate_response = dspy.ChainOfThought(
"intent, sentiment, knowledge, company_policy -> response, action_needed"
)
self.escalate = dspy.Predict("issue, customer_details -> escalation_reason, department")
def forward(self, customer_message):
# 1. Classify Intent
classification = self.classify_intent(customer_message=customer_message)
# 2. Retrieve Knowledge
relevant_kb = self.search_knowledge(
query=f"{classification.intent} {customer_message}"
)
# 3. Generate Response
response = self.generate_response(
intent=classification.intent,
sentiment=classification.sentiment,
knowledge="\n".join(relevant_kb.passages),
company_policy=self.knowledge_base.get("policies", "")
)
# 4. Check for Escalation
final_action = response.action_needed
if "escalate" in str(response.action_needed).lower():
escalation = self.escalate(
issue=customer_message,
customer_details="Session Context..."
)
final_action = f"Escalated to {escalation.department}"
return dspy.Prediction(
response=response.response,
action=final_action
)Best Practices
- Clear Goals: Define explicit goals and alignment checks for agents.
- Error Handling: Implement robust fallback mechanisms for failed actions or API calls.
- Continuous Learning: Allow agents to store feedback and adjust strategies over time (e.g., using a memory module).
- Safety: Constrain agent actions to prevent unintended consequences.