DSPy Integration

pip install mantisdk dspy-ai

from dotenv import load_dotenv
load_dotenv()
 
# Initialize tracing FIRST
import mantisdk.tracing as tracing
 
tracing.init(service_name="dspy-agent", instrument_default=False)
tracing.instrument(names=["dspy"])
 
# NOW import DSPy
import dspy
 
# Configure DSPy
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)
 
# Define signature
class QA(dspy.Signature):
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()
 
# Use module - automatically traced
cot = dspy.ChainOfThought(QA)
result = cot(question="What is the capital of France?")
print(result.answer)
 
# Clean up
tracing.shutdown()

import mantisdk.tracing as tracing
import dspy
 
@tracing.trace
def classify_and_answer(query: str) -> dict:
    """Decorator auto-captures function I/O."""
    # Classify
    query_type = classify_query(query)
 
    # ChainOfThought call is auto-instrumented
    cot = dspy.ChainOfThought(QA)
    result = cot(question=query)
 
    return {
        "type": query_type,
        "answer": result.answer,
        "reasoning": result.reasoning,
    }

#!/usr/bin/env python3
"""DSPy Reasoning Agent with Mantis SDK Tracing."""
 
import argparse
from dotenv import load_dotenv
load_dotenv()
 
# Initialize tracing FIRST
import mantisdk.tracing as tracing
from mantisdk.tracing.exporters.insight import is_insight_configured
 
def setup_tracing(trace_to_console: bool = False):
    """Initialize mantisdk tracing with DSPy instrumentation."""
    exporters = []
 
    if is_insight_configured():
        exporters.append(tracing.insight_exporter())
        print("[Tracing] Insight exporter configured")
    else:
        print("[Tracing] No Insight credentials found")
 
    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())
 
    tracing.init(
        service_name="dspy-reasoning-agent",
        exporters=exporters if exporters else None,
        instrument_default=False,
    )
 
    tracing.instrument(names=["dspy"])
    print("[Tracing] DSPy instrumentation enabled")
 
 
# NOW import DSPy
import dspy
 
 
# Define signatures
class QuestionAnswer(dspy.Signature):
    """Answer a question with clear reasoning."""
    question: str = dspy.InputField(desc="The question to answer")
    reasoning: str = dspy.OutputField(desc="Step-by-step reasoning")
    answer: str = dspy.OutputField(desc="The final answer")
 
 
class MathProblem(dspy.Signature):
    """Solve a math problem step by step."""
    problem: str = dspy.InputField(desc="The math problem")
    steps: str = dspy.OutputField(desc="Solution steps")
    final_answer: str = dspy.OutputField(desc="The numerical answer")
 
 
# Agent class with manual tracing
class ReasoningAgent:
    """DSPy-based reasoning agent with tracing."""
 
    def __init__(self, model: str = "gpt-4o-mini"):
        self.lm = dspy.LM(f"openai/{model}", temperature=0.7)
        dspy.configure(lm=self.lm)
 
        self.qa_module = dspy.ChainOfThought(QuestionAnswer)
        self.math_module = dspy.ChainOfThought(MathProblem)
 
    @tracing.trace(name="classify_task")
    def _classify_task(self, query: str) -> str:
        """Classify the type of reasoning task."""
        query_lower = query.lower()
        math_keywords = ["calculate", "compute", "solve", "%", "+", "-"]
 
        if any(kw in query_lower for kw in math_keywords):
            if any(c.isdigit() for c in query):
                return "math"
        return "general"
 
    @tracing.trace(name="reason_general")
    def _reason_general(self, question: str) -> dict:
        """General Q&A reasoning - auto-traced."""
        result = self.qa_module(question=question)
        return {
            "type": "general",
            "reasoning": result.reasoning,
            "answer": result.answer,
        }
 
    @tracing.trace(name="reason_math")
    def _reason_math(self, problem: str) -> dict:
        """Math reasoning - auto-traced."""
        result = self.math_module(problem=problem)
        return {
            "type": "math",
            "steps": result.steps,
            "answer": result.final_answer,
        }
 
    def reason(self, query: str) -> str:
        """Process a query with appropriate reasoning."""
        with tracing.trace("reasoning_task", input=query) as span:
            task_type = self._classify_task(query)
            span.set_attribute("task_type", task_type)
 
            if task_type == "math":
                result = self._reason_math(query)
                response = f"Steps:\n{result['steps']}\n\nAnswer: {result['answer']}"
            else:
                result = self._reason_general(query)
                response = f"Reasoning:\n{result['reasoning']}\n\nAnswer: {result['answer']}"
 
            span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
            return response
 
 
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--query", "-q", default="What is 15% of 240?")
    parser.add_argument("--trace-to-console", action="store_true")
    args = parser.parse_args()
 
    setup_tracing(trace_to_console=args.trace_to_console)
 
    agent = ReasoningAgent()
    result = agent.reason(args.query)
    print(result)
 
    tracing.shutdown()
 
 
if __name__ == "__main__":
    main()

reasoning_task (manual trace)
├── classify_task (manual span)
└── reason_math (manual span)
    └── ChainOfThought.forward (auto-instrumented)
        └── LM.generate (auto-instrumented)
            ├── model: gpt-4o-mini
            ├── input_tokens: 45
            └── output_tokens: 120

# OpenAI
lm = dspy.LM("openai/gpt-4o-mini")
 
# Anthropic
lm = dspy.LM("anthropic/claude-sonnet-4-20250514")
 
# Local model via LiteLLM
lm = dspy.LM("ollama/llama2")

lm = dspy.LM(
    "openai/gpt-4o-mini",
    temperature=0.7,
    max_tokens=1000,
)