DSPy Reasoning Agent

pip install mantisdk dspy-ai python-dotenv

# .env file
INSIGHT_HOST=https://insight.withmetis.ai
INSIGHT_PUBLIC_KEY=pk-lf-your-key
INSIGHT_SECRET_KEY=sk-lf-your-key
OPENAI_API_KEY=sk-your-openai-key

#!/usr/bin/env python3
"""DSPy Reasoning Agent with Mantis SDK Tracing."""
 
import argparse
import asyncio
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. "
              "Set INSIGHT_HOST, INSIGHT_PUBLIC_KEY, INSIGHT_SECRET_KEY")
 
    if trace_to_console:
        from opentelemetry.sdk.trace.export import ConsoleSpanExporter
        exporters.append(ConsoleSpanExporter())
        print("[Tracing] Console exporter enabled")
 
    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 (after tracing init)
import dspy
 
 
# Define DSPy 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 process")
    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 to solve")
    steps: str = dspy.OutputField(desc="Step-by-step solution process")
    final_answer: str = dspy.OutputField(desc="The numerical answer")
 
 
class AnalyzeArgument(dspy.Signature):
    """Analyze an argument for logical validity."""
    argument: str = dspy.InputField(desc="The argument to analyze")
    analysis: str = dspy.OutputField(desc="Analysis of the argument's logic")
    conclusion: str = dspy.OutputField(desc="Whether the argument is valid/invalid")
 
 
class ReasoningAgent:
    """DSPy-based reasoning agent with chain-of-thought."""
 
    def __init__(self, model: str = "gpt-4o-mini"):
        # Configure DSPy
        self.lm = dspy.LM(f"openai/{model}", temperature=0.7)
        dspy.configure(lm=self.lm)
 
        # Create reasoning modules
        self.qa_module = dspy.ChainOfThought(QuestionAnswer)
        self.math_module = dspy.ChainOfThought(MathProblem)
        self.argument_module = dspy.ChainOfThought(AnalyzeArgument)
 
        self.conversation_history = []
 
    @tracing.trace(name="classify_task")
    def _classify_task(self, query: str) -> str:
        """Classify the type of reasoning task.
 
        Using @tracing.trace decorator - automatically captures input/output.
        """
        query_lower = query.lower()
 
        math_keywords = ["calculate", "compute", "solve", "what is",
                        "how much", "%", "+", "-", "*", "/", "="]
        argument_keywords = ["argue", "argument", "therefore", "because",
                            "valid", "invalid", "premise", "conclude"]
 
        if any(kw in query_lower for kw in math_keywords):
            if any(c.isdigit() for c in query):
                return "math"
 
        if any(kw in query_lower for kw in argument_keywords):
            return "argument"
 
        return "general"
 
    @tracing.trace(name="reason_general")
    def _reason_general(self, question: str) -> dict:
        """General Q&A reasoning - input/output auto-captured."""
        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:
        """Mathematical reasoning - input/output auto-captured."""
        result = self.math_module(problem=problem)
        return {
            "type": "math",
            "steps": result.steps,
            "answer": result.final_answer
        }
 
    @tracing.trace(name="reason_argument")
    def _reason_argument(self, argument: str) -> dict:
        """Argument analysis - input/output auto-captured."""
        result = self.argument_module(argument=argument)
        return {
            "type": "argument",
            "analysis": result.analysis,
            "conclusion": result.conclusion
        }
 
    async def reason(self, query: str) -> str:
        """Process a query with appropriate reasoning strategy.
 
        Using trace context manager with explicit input/output capture.
        """
        with tracing.trace("reasoning_task", input=query) as span:
            # Classify the task
            task_type = self._classify_task(query)
            span.set_attribute("task_type", task_type)
 
            # Execute appropriate reasoning
            if task_type == "math":
                result = self._reason_math(query)
                response = f"**Solution Steps:**\n{result['steps']}\n\n**Answer:** {result['answer']}"
            elif task_type == "argument":
                result = self._reason_argument(query)
                response = f"**Analysis:**\n{result['analysis']}\n\n**Conclusion:** {result['conclusion']}"
            else:
                result = self._reason_general(query)
                response = f"**Reasoning:**\n{result['reasoning']}\n\n**Answer:** {result['answer']}"
 
            # Store in history
            self.conversation_history.append({
                "query": query,
                "type": task_type,
                "response": response
            })
 
            # Set output using semantic conventions
            span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)
            return response
 
    def reset(self):
        """Reset conversation history."""
        self.conversation_history = []
 
 
async def run_interactive_session(agent: ReasoningAgent):
    """Run an interactive CLI session."""
    print("\n" + "="*60)
    print("  DSPy Reasoning Agent (with Mantis Tracing)")
    print("="*60)
    print("\nI can help you with:")
    print("  - General questions (chain-of-thought reasoning)")
    print("  - Math problems (step-by-step solutions)")
    print("  - Argument analysis (logical validity)")
    print("\nType 'quit' or 'exit' to end the session.")
    print("="*60 + "\n")
 
    while True:
        try:
            user_input = input("You: ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\n\nGoodbye!")
            break
 
        if not user_input:
            continue
 
        if user_input.lower() in ("quit", "exit", "q"):
            print("\nGoodbye!")
            break
 
        if user_input.lower() == "clear":
            agent.reset()
            print("Conversation history cleared.\n")
            continue
 
        print("\nAssistant:\n")
        try:
            response = await agent.reason(user_input)
            print(response)
        except Exception as e:
            print(f"[Error] {e}")
        print()
 
 
async def run_single_query(agent: ReasoningAgent, query: str):
    """Run a single query."""
    try:
        response = await agent.reason(query)
        print(response)
    except Exception as e:
        print(f"[Error] {e}")
 
 
def main():
    parser = argparse.ArgumentParser(description="DSPy Reasoning Agent")
    parser.add_argument("--model", default="gpt-4o-mini")
    parser.add_argument("--query", "-q", help="Single query to run")
    parser.add_argument("--trace-to-console", action="store_true")
    parser.add_argument("--no-tracing", action="store_true")
    args = parser.parse_args()
 
    if not args.no_tracing:
        setup_tracing(trace_to_console=args.trace_to_console)
 
    agent = ReasoningAgent(model=args.model)
 
    try:
        if args.query:
            asyncio.run(run_single_query(agent, args.query))
        else:
            asyncio.run(run_interactive_session(agent))
    finally:
        if not args.no_tracing:
            tracing.shutdown()
 
 
if __name__ == "__main__":
    main()

python reasoning_agent.py

python reasoning_agent.py --query "What is 15% of 240?"

python reasoning_agent.py --trace-to-console

reasoning_task
├── classify_task
│   ├── input: "What is 15% of 240?"
│   └── output: "math"
└── reason_math
    └── ChainOfThought.forward (auto-instrumented)
        └── LM.generate
            ├── model: gpt-4o-mini
            ├── input_tokens: 45
            └── output_tokens: 120

@tracing.trace(name="classify_task")
def _classify_task(self, query: str) -> str:
    # Input/output automatically captured
    return task_type

with tracing.trace("reasoning_task", input=query) as span:
    span.set_attribute("task_type", task_type)
    result = process(query)
    span.set_attribute(tracing.semconv.TRACE_OUTPUT, response)