DSPy Integration Guide

pip install dspy

pip install git+https://github.com/stanfordnlp/dspy.git

import os
import dspy

# Set environment variables for APIpie
os.environ["OPENAI_API_KEY"] = "your-apipie-api-key"
os.environ["OPENAI_API_BASE"] = "https://apipie.ai/v1"

# Configure DSPy to use APIpie
lm = dspy.OpenAI(model="gpt-4o-mini")  # Use any model available on APIpie
dspy.configure(lm=lm)

import dspy

# Configure DSPy with explicit parameters
lm = dspy.OpenAI(
    model="gpt-4o-mini",  # Use any model available on APIpie
    api_key="your-apipie-api-key",
    api_base="https://apipie.ai/v1"
)

# Set this as the default language model
dspy.configure(lm=lm)

import os
import dspy

# Configure DSPy with APIpie
os.environ["OPENAI_API_KEY"] = "your-apipie-api-key"
os.environ["OPENAI_API_BASE"] = "https://apipie.ai/v1"

# Set up the language model
lm = dspy.OpenAI(model="gpt-4o-mini")
dspy.configure(lm=lm)

# Define a simple question-answering module
class BasicQA(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate_answer = dspy.ChainOfThought("question -> answer")

    def forward(self, question):
        return self.generate_answer(question=question)

# Create and use the QA module
qa_module = BasicQA()
response = qa_module("What is the capital of France?")

print(response.answer)  # Paris

import os
import dspy
from typing import List

# Configure DSPy with APIpie
os.environ["OPENAI_API_KEY"] = "your-apipie-api-key"
os.environ["OPENAI_API_BASE"] = "https://apipie.ai/v1"

# Set up the language model
lm = dspy.OpenAI(model="gpt-4o")
dspy.configure(lm=lm)

# Define a tool for math operations
def evaluate_math(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return float(eval(expression))

# Define a simple information retrieval tool
def search_web(query: str) -> List[str]:
    """Search the web for information."""
    # In a real application, this would use a search API
    # This is just a mock example
    if "Paris" in query:
        return ["Paris is the capital of France.", "Paris has a population of about 2.2 million."]
    elif "Rome" in query:
        return ["Rome is the capital of Italy.", "Rome was founded in 753 BC."]
    else:
        return ["No specific information found."]

# Create a ReAct agent that can use these tools
react_agent = dspy.ReAct(
    "question -> answer",
    tools=[evaluate_math, search_web]
)

# Use the agent to answer questions
result1 = react_agent(question="What is 123 * 456?")
print(f"Math answer: {result1.answer}")

result2 = react_agent(question="What is the capital of Italy?")
print(f"Information answer: {result2.answer}")

import os
import dspy
from dspy.retrieve import ColBERTv2

# Configure DSPy with APIpie
os.environ["OPENAI_API_KEY"] = "your-apipie-api-key"
os.environ["OPENAI_API_BASE"] = "https://apipie.ai/v1"

# Set up the language model
lm = dspy.OpenAI(model="gpt-4o-mini")
dspy.configure(lm=lm)

# Mock retriever for demonstration
# In a real application, you would use a proper retrieval system
# For example: retriever = ColBERTv2(url="http://your-colbert-server")
class MockRetriever:
    def __call__(self, query, k=3):
        # Simulate document retrieval
        if "climate change" in query.lower():
            docs = [
                {"text": "Climate change is a significant global challenge that requires immediate action."},
                {"text": "Rising temperatures are leading to more extreme weather events worldwide."},
                {"text": "Renewable energy is a key solution to address climate change."}
            ]
        else:
            docs = [
                {"text": "No specific information found for this query."}
            ]
        return docs[:k]

retriever = MockRetriever()

# Define a RAG module using DSPy
class RAG(dspy.Module):
    def __init__(self, retriever):
        super().__init__()
        self.retriever = retriever
        self.generate_query = dspy.Predict("question -> query")
        self.generate_answer = dspy.ChainOfThought("question, context -> answer")

    def forward(self, question):
        # Generate an effective query
        query = self.generate_query(question=question).query

        # Retrieve relevant documents
        docs = self.retriever(query)
        context = "\n".join([d["text"] for d in docs])

        # Generate an answer based on the retrieved information
        return self.generate_answer(question=question, context=context)

# Create the RAG system
rag_system = RAG(retriever)

# Optimize the RAG system
teleprompter = dspy.Teleprompter(introspect=True)
optimized_rag = teleprompter.optimize(
    rag_system,
    task_demos=[
        dspy.Example(
            question="What are the main causes of climate change?",
            answer="The main causes of climate change include greenhouse gas emissions from burning fossil fuels, deforestation, and industrial processes."
        ),
        dspy.Example(
            question="How can we address climate change?",
            answer="Climate change can be addressed through reducing greenhouse gas emissions, transitioning to renewable energy, improving energy efficiency, and implementing sustainable practices."
        )
    ]
)

# Use the optimized RAG system
result = optimized_rag("What are the effects of climate change on agriculture?")
print(result.answer)

import os
import dspy

# Configure DSPy with APIpie
os.environ["OPENAI_API_KEY"] = "your-apipie-api-key"
os.environ["OPENAI_API_BASE"] = "https://apipie.ai/v1"

# Set up the language model
lm = dspy.OpenAI(model="gpt-4o")
dspy.configure(lm=lm)

# Define signatures for multi-step reasoning
class GeneratePlan(dspy.Signature):
    """Generate a step-by-step plan to solve a complex problem."""
    problem = dspy.InputField()
    plan = dspy.OutputField(desc="A detailed step-by-step plan with 3-5 steps")

class ExecuteStep(dspy.Signature):
    """Execute a specific step in the problem-solving plan."""
    problem = dspy.InputField()
    previous_steps = dspy.InputField(desc="Results from previous steps, if any")
    current_step = dspy.InputField(desc="The current step to execute")
    result = dspy.OutputField(desc="The result of executing the current step")

class ProvideFinalAnswer(dspy.Signature):
    """Provide the final answer based on all steps executed."""
    problem = dspy.InputField()
    all_steps = dspy.InputField(desc="All steps and their results")
    answer = dspy.OutputField(desc="The final answer to the problem")

# Create a multi-step reasoning module
class MultistepReasoning(dspy.Module):
    def __init__(self):
        super().__init__()
        self.generate_plan = dspy.ChainOfThought(GeneratePlan)
        self.execute_step = dspy.ChainOfThought(ExecuteStep)
        self.provide_final_answer = dspy.ChainOfThought(ProvideFinalAnswer)

    def forward(self, problem):
        # Generate a plan
        plan_output = self.generate_plan(problem=problem)

        # Parse the plan into steps
        steps = [step.strip() for step in plan_output.plan.split('\n') if step.strip()]

        # Execute each step
        previous_results = []
        all_results = []

        for i, step in enumerate(steps):
            # Execute the current step
            step_result = self.execute_step(
                problem=problem,
                previous_steps='\n'.join(previous_results),
                current_step=step
            )

            # Store the result
            result_text = f"Step {i+1}: {step}\nResult: {step_result.result}"
            previous_results.append(result_text)
            all_results.append(result_text)

        # Provide the final answer
        final_answer = self.provide_final_answer(
            problem=problem,
            all_steps='\n'.join(all_results)
        )

        return final_answer.answer

# Use the multi-step reasoning module
reasoning_module = MultistepReasoning()
answer = reasoning_module("Calculate the approximate area of a circle with diameter 10 cm.")
print(answer)