From dspy-skills
Builds production-quality DSPy ReAct agents for multi-step tasks using tools, with reasoning, acting, and error handling.
npx claudepluginhub omidzamani/dspy-skillsThis skill is limited to using the following tools:
Build production-quality ReAct agents that use tools to solve complex multi-step tasks with reasoning, acting, and error handling.
Optimizes complex DSPy agentic systems using GEPA with LLM reflection on execution traces, textual feedback metrics, and Pareto-based evolutionary search.
Designs and optimizes AI agent action spaces, tool definitions, observation formats, error recovery, and context for higher task completion rates.
Designs and optimizes AI agent action spaces, tool definitions, observation formatting, error recovery, and context budgeting for higher task completion rates.
Share bugs, ideas, or general feedback.
Build production-quality ReAct agents that use tools to solve complex multi-step tasks with reasoning, acting, and error handling.
| Input | Type | Description |
|---|---|---|
signature | str | Task signature (e.g., "question -> answer") |
tools | list[callable] | Available tools/functions |
max_iters | int | Max reasoning steps (default: 20) |
| Output | Type | Description |
|---|---|---|
agent | dspy.ReAct | Configured ReAct agent |
Tools are Python functions with clear docstrings. The agent uses docstrings to understand tool capabilities:
import dspy
def search(query: str) -> list[str]:
"""Search knowledge base for relevant information.
Args:
query: Search query string
Returns:
List of relevant text passages
"""
retriever = dspy.ColBERTv2(url='http://20.102.90.50:2017/wiki17_abstracts')
results = retriever(query, k=3)
return [r['text'] for r in results]
def calculate(expression: str) -> float:
"""Safely evaluate mathematical expressions.
Args:
expression: Math expression (e.g., "2 + 2", "sqrt(16)")
Returns:
Numerical result
"""
try:
interpreter = dspy.PythonInterpreter()
return interpreter.execute(expression)
except Exception as e:
return f"Error: {e}"
# Configure LM
dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))
# Create agent
agent = dspy.ReAct(
signature="question -> answer",
tools=[search, calculate],
max_iters=5
)
# Use agent
result = agent(question="What is the population of Paris plus 1000?")
print(result.answer)
import dspy
import logging
logger = logging.getLogger(__name__)
class ResearchAgent(dspy.Module):
"""Production agent with error handling and logging."""
def __init__(self, max_iters: int = 5):
self.max_iters = max_iters
self.agent = dspy.ReAct(
signature="question -> answer",
tools=[self.search, self.calculate, self.summarize],
max_iters=max_iters
)
def search(self, query: str) -> list[str]:
"""Search for relevant documents."""
try:
retriever = dspy.ColBERTv2(
url='http://20.102.90.50:2017/wiki17_abstracts'
)
results = retriever(query, k=5)
return [r['text'] for r in results]
except Exception as e:
logger.error(f"Search failed: {e}")
return [f"Search unavailable: {e}"]
def calculate(self, expression: str) -> str:
"""Evaluate mathematical expressions safely."""
try:
interpreter = dspy.PythonInterpreter()
result = interpreter.execute(expression)
return str(result)
except Exception as e:
logger.error(f"Calculation failed: {e}")
return f"Error: {e}"
def summarize(self, text: str) -> str:
"""Summarize long text into key points."""
try:
summarizer = dspy.Predict("text -> summary: str")
return summarizer(text=text[:1000]).summary
except Exception as e:
logger.error(f"Summarization failed: {e}")
return "Summarization unavailable"
def forward(self, question: str) -> dspy.Prediction:
"""Execute agent with error handling."""
try:
return self.agent(question=question)
except Exception as e:
logger.error(f"Agent failed: {e}")
return dspy.Prediction(answer=f"Error: {e}")
# Usage
agent = ResearchAgent(max_iters=6)
response = agent(question="What is the capital of France and its population?")
print(response.answer)
ReAct agents benefit from reflective optimization:
from dspy.evaluate import Evaluate
def feedback_metric(example, pred, trace=None, pred_name=None, pred_trace=None):
"""Provide textual feedback for GEPA."""
is_correct = example.answer.lower() in pred.answer.lower()
score = 1.0 if is_correct else 0.0
feedback = "Correct." if is_correct else f"Expected '{example.answer}'. Check tool selection."
return dspy.Prediction(score=score, feedback=feedback)
# Optimize agent
optimizer = dspy.GEPA(
metric=feedback_metric,
reflection_lm=dspy.LM("openai/gpt-4o"),
auto="medium",
enable_tool_optimization=True # Also optimize tool docstrings
)
compiled = optimizer.compile(agent, trainset=trainset)
compiled.save("research_agent_optimized.json", save_program=False)
max_iters to prevent infinite loops (default is 20, but 5-10 often sufficient for simpler tasks)