Loading...
Loading...
Compare original and translation side by side
$ARGUMENTS/
├── main.py # Entry point — run your AI feature
├── program.py # AI logic (DSPy module)
├── metrics.py # How to measure if the AI is working
├── optimize.py # Make the AI better automatically
├── evaluate.py # Test the AI's quality
├── data.py # Training/test data loading
└── requirements.txt # Dependencies$ARGUMENTS/
├── main.py # Entry point — run your AI feature
├── program.py # AI logic (DSPy module)
├── metrics.py # How to measure if the AI is working
├── optimize.py # Make the AI better automatically
├── evaluate.py # Test the AI's quality
├── data.py # Training/test data loading
└── requirements.txt # Dependenciesrequirements.txtrequirements.txtdspy>=2.5datasetsdspy>=2.5datasetsdata.pydata.pyimport dspy
def load_data():
"""Load and prepare training/dev data.
Returns:
tuple: (trainset, devset) as lists of dspy.Example
"""
# TODO: Replace with actual data loading
examples = [
dspy.Example(input_field="...", output_field="...").with_inputs("input_field"),
]
split = int(0.8 * len(examples))
return examples[:split], examples[split:]import dspy
def load_data():
"""Load and prepare training/dev data.
Returns:
tuple: (trainset, devset) as lists of dspy.Example
"""
# TODO: Replace with actual data loading
examples = [
dspy.Example(input_field="...", output_field="...").with_inputs("input_field"),
]
split = int(0.8 * len(examples))
return examples[:split], examples[split:]program.pyprogram.pydspy.Predictdspy.ChainOfThoughtdspy.ProgramOfThoughtdspy.ReActimport dspy
class MySignature(dspy.Signature):
"""Describe the task here."""
# Adapt fields to user's task
input_field: str = dspy.InputField(desc="description")
output_field: str = dspy.OutputField(desc="description")
class MyProgram(dspy.Module):
def __init__(self):
self.predict = dspy.ChainOfThought(MySignature)
def forward(self, **kwargs):
return self.predict(**kwargs)dspy.Predictdspy.ChainOfThoughtdspy.ProgramOfThoughtdspy.ReActimport dspy
class MySignature(dspy.Signature):
"""Describe the task here."""
# Adapt fields to user's task
input_field: str = dspy.InputField(desc="description")
output_field: str = dspy.OutputField(desc="description")
class MyProgram(dspy.Module):
def __init__(self):
self.predict = dspy.ChainOfThought(MySignature)
def forward(self, **kwargs):
return self.predict(**kwargs)metrics.pymetrics.pydef metric(example, prediction, trace=None):
"""Score how good the AI's output is.
Args:
example: Expected output (ground truth)
prediction: What the AI actually produced
trace: Optional trace for optimization
Returns:
float: Score between 0 and 1
"""
# TODO: Implement task-specific metric
return prediction.output_field == example.output_fielddef metric(example, prediction, trace=None):
"""Score how good the AI's output is.
Args:
example: Expected output (ground truth)
prediction: What the AI actually produced
trace: Optional trace for optimization
Returns:
float: Score between 0 and 1
"""
# TODO: Implement task-specific metric
return prediction.output_field == example.output_fieldevaluate.pyevaluate.pyimport dspy
from dspy.evaluate import Evaluate
from program import MyProgram
from metrics import metric
from data import load_dataimport dspy
from dspy.evaluate import Evaluate
from program import MyProgram
from metrics import metric
from data import load_dataundefinedundefinedoptimize.pyoptimize.pyimport dspy
from program import MyProgram
from metrics import metric
from data import load_dataimport dspy
from program import MyProgram
from metrics import metric
from data import load_dataundefinedundefinedmain.pymain.pyimport dspy
from program import MyProgramimport dspy
from program import MyProgramundefinedundefined$ARGUMENTS/
├── main.py # Entry point — run your AI feature
├── program.py # AI logic (DSPy module)
├── server.py # FastAPI app — routes and startup
├── models.py # Pydantic request/response schemas
├── config.py # Environment configuration
├── metrics.py # How to measure if the AI is working
├── optimize.py # Make the AI better automatically
├── evaluate.py # Test the AI's quality
├── data.py # Training/test data loading
├── requirements.txt # Dependencies
├── Dockerfile
└── .env.example$ARGUMENTS/
├── main.py # Entry point — run your AI feature
├── program.py # AI logic (DSPy module)
├── server.py # FastAPI app — routes and startup
├── models.py # Pydantic request/response schemas
├── config.py # Environment configuration
├── metrics.py # How to measure if the AI is working
├── optimize.py # Make the AI better automatically
├── evaluate.py # Test the AI's quality
├── data.py # Training/test data loading
├── requirements.txt # Dependencies
├── Dockerfile
└── .env.exampleserver.pyserver.pyfrom contextlib import asynccontextmanager
import dspy
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from program import MyProgram
@asynccontextmanager
async def lifespan(app: FastAPI):
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)
app.state.program = MyProgram()
try:
app.state.program.load("optimized.json")
except FileNotFoundError:
pass
yield
app = FastAPI(title="My AI API", lifespan=lifespan)
class QueryRequest(BaseModel):
input_field: str = Field(..., min_length=1)
class QueryResponse(BaseModel):
output_field: str
@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
result = app.state.program(input_field=request.input_field)
return QueryResponse(output_field=result.output_field)
@app.get("/health")
async def health():
return {"status": "ok"}QueryRequestQueryResponsefrom contextlib import asynccontextmanager
import dspy
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from program import MyProgram
@asynccontextmanager
async def lifespan(app: FastAPI):
lm = dspy.LM("openai/gpt-4o-mini")
dspy.configure(lm=lm)
app.state.program = MyProgram()
try:
app.state.program.load("optimized.json")
except FileNotFoundError:
pass
yield
app = FastAPI(title="My AI API", lifespan=lifespan)
class QueryRequest(BaseModel):
input_field: str = Field(..., min_length=1)
class QueryResponse(BaseModel):
output_field: str
@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
result = app.state.program(input_field=request.input_field)
return QueryResponse(output_field=result.output_field)
@app.get("/health")
async def health():
return {"status": "ok"}QueryRequestQueryResponserequirements.txtrequirements.txtdspy>=2.5
fastapi>=0.100
uvicorn[standard]
pydantic-settings>=2.0dspy>=2.5
fastapi>=0.100
uvicorn[standard]
pydantic-settings>=2.0.env.example.env.exampleAI_MODEL_NAME=openai/gpt-4o-mini
AI_API_KEY=your-api-key-hereAI_MODEL_NAME=openai/gpt-4o-mini
AI_API_KEY=your-api-key-heredata.py/ai-generating-dataevaluate.pyoptimize.pymain.py/ai-serving-apis/ai-improving-accuracydata.py/ai-generating-dataevaluate.pyoptimize.pymain.py/ai-serving-apis/ai-improving-accuracy