Reliable, optimizable LLM steps with zero DSPy boilerplate: typed outputs, automatic self-correction, and one-call prompt tuning.

pip install dspyer
# or using uv:
uv add dspyer

pip install git+https://github.com/theramkm/dspyer.git
# or using uv:
uv add git+https://github.com/theramkm/dspyer.git

import dspy
from pydantic import BaseModel, Field, field_validator
from dspyer import AgentTranspiler, Graph, MockCompletionResult, StatefulNode

# 1. Describe the schema contract you want the LLM to honor
class Query(BaseModel):
    query: str

class RAGResponse(BaseModel):
    answer: str = Field(description="Answer referencing the sources")
    citations: list[str] = Field(description="Sources cited, e.g. ['doc_1']")

    @field_validator("citations")
    @classmethod
    def must_cite(cls, v):
        if not v:  # Ensure we cite at least one source
            raise ValueError("Answer must cite at least one source.")
        return v

# 2. Define an optimizable, self-correcting node
node = StatefulNode(
    "Synthesizer", Query, RAGResponse,
    instructions="Answer the query and cite sources.",
    max_retries=3,
)
graph = Graph()
graph.add_node(node)
graph.set_entry_point("Synthesizer")
program = AgentTranspiler.compile(graph)

# 3. Offline mock: configuration and run
# (Hiding MockLM details for readability; click below to expand)

r = program(query="What license is dspyer under?")

print("Answer:   ", r.answer)                                   # Apache-2.0 [doc_1].
print("Citations:", r.citations)                                # ['doc_1']
print("Self-correction loops:", r["_metadata"]["refinement_steps_taken"])  # 1

from dspyer import self_correcting
from pydantic import BaseModel

class SolverOutput(BaseModel):
    answer: str
    steps: list[str]

# Both synchronous (def) and asynchronous (async def) functions are fully supported!
@self_correcting(max_retries=3)
async def solve(question: str) -> SolverOutput:
    """Answer the question and outline the logic steps."""
    # Body is intentionally empty; dspyer generates the call from the signature
    pass

# Await the call naturally in async environments:
result = await solve(question="What is the capital of France?")

@self_correcting(schema=SolverOutput, max_retries=3)
class Solver(dspy.Module):
    def __init__(self):
        super().__init__()
        self.solve = dspy.Predict("question -> answer, steps")

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

from dspy.teleprompt import BootstrapFewShot

def metric(example, pred, trace=None) -> bool:
    return example.sentiment.lower() == pred.sentiment.lower()

optimizer = BootstrapFewShot(metric=metric, max_bootstrapped_demos=2)
optimized = optimizer.compile(program, trainset=trainset)

# Save prompts
optimized.save_prompts("agent_config.json")

# Load in production
production_program.load_prompts("agent_config.json")

compiled_agent = AgentTranspiler.compile(graph)

def run_agent_node(state):
    pred = compiled_agent(query=state["user_query"])
    return {"agent_response": pred.answer, "citations": pred.citations}

from dspyer import from_langgraph

node_mappings = {
    "Clean": StatefulNode("Clean", CleanInput, CleanOutput, instructions="Normalize the query"),
    "Solve": StatefulNode("Solve", SolveInput, SolveOutput, instructions="Answer the query"),
}
graph = from_langgraph(builder, node_mappings=node_mappings)
program = AgentTranspiler.compile(graph)

program = AgentTranspiler.compile(graph, validation_log_path="logs/validation.jsonl")

from dspyer import generate_validation_report

print(generate_validation_report("logs/validation.jsonl"))

==================================================
           dspyer Batch Validation Report
==================================================

Node: Synthesizer
--------------------------------------------------
  Total Runs: 10
  Successful Runs: 8 (80.0%)
  Failed Runs: 2 (20.0%)
  Retry Rate: 40.0% (4/10 runs required retries)
  Average Retries: 0.80 per run
  Top Failing Pydantic Fields:
    - citations: 4 errors (66.7% of total errors)
    - answer: 2 errors (33.3% of total errors)

==================================================

program = AgentTranspiler.compile(graph, dataset_log_path="logs/flywheel.jsonl")

from dspyer import load_logged_dataset

# We must specify which keys act as model inputs
trainset = load_logged_dataset(
    dataset_log_path="logs/flywheel.jsonl",
    input_keys=["query"]
)

from dspyer import dspyer_node

class ExtractorInput(BaseModel):
    query: str

class ExtractorOutput(BaseModel):
    entities: list[str]

@dspyer_node(
    input_model=ExtractorInput,
    output_model=ExtractorOutput,
    instructions="Extract named entities from the user query."
)
def extract_entities_node(state):
    # This node is explicitly registered with its typing contract
    # Bypasses AST static analysis during LangGraph conversion
    pass

program = AgentTranspiler.compile(graph, output_model=ExtractorOutput)

# 1. Async forward call
result = await program.aforward(query="Alice and Bob went to Paris")
print(result.entities)

# 2. Async event streaming
async for event in program.astream(query="Alice and Bob went to Paris"):
    print(f"Event: {event['event']} | Node: {event.get('node')}")

from dspyer import BaseStorageAdapter, set_storage_adapter

class CustomDatabaseAdapter(BaseStorageAdapter):
    def append_line(self, target: str, line: str) -> None:
        # Custom synchronous DB write
        db.insert(target, line)

    async def append_line_async(self, target: str, line: str) -> None:
        # Custom non-blocking async DB write
        await db.async_insert(target, line)

# Register custom adapter globally
set_storage_adapter(CustomDatabaseAdapter())