Reward Servers

from adaptive_sdk.external import RewardServer, ValidatedRequest, Response, ServerInfo
from pydantic import BaseModel, Field

# Define metadata schema (validated against dataset records)
class ScaryLetterMetadata(BaseModel):
    scary_letter: str = Field(min_length=1, max_length=1)

class ScaryLetterRewardServer(RewardServer[ScaryLetterMetadata]):
    def __init__(self, port: int = 8000, blocking: bool = True, **kwargs):
        super().__init__(port, ScaryLetterMetadata, blocking, **kwargs)

    async def score(self, request: ValidatedRequest[ScaryLetterMetadata]) -> Response:
        last_completion = request.turns[-1].content
        has_scary = request.metadata.scary_letter in last_completion
        return Response(
            reward=0.0 if has_scary else 1.0,
            metadata={"feedback": "No scary letters!" if not has_scary else "Found scary letters!"}
        )

    async def info(self) -> ServerInfo:
        return ServerInfo(
            version="1.0",
            name="Scary Letter Detector",
            description="Rewards completions that avoid a specific letter"
        )

if __name__ == "__main__":
    server = ScaryLetterRewardServer(port=50056)

from adaptive_sdk.external import RewardServer, EmptyMetadata, ValidatedRequest, Response, ServerInfo

class LengthRewardServer(RewardServer[EmptyMetadata]):
    def __init__(self, port: int = 8000, blocking: bool = True, **kwargs):
        super().__init__(port, EmptyMetadata, blocking, **kwargs)

    async def score(self, request: ValidatedRequest[EmptyMetadata]) -> Response:
        completion = request.turns[-1].content
        return Response(reward=min(len(completion) / 100, 1.0), metadata={})

    async def info(self) -> ServerInfo:
        return ServerInfo(version="1.0", name="Length Rewarder", description="Rewards longer responses")

if __name__ == "__main__":
    server = LengthRewardServer(port=50056)

import sqlite3
import os
import pandas as pd
from adaptive_sdk.external import RewardServer, ValidatedRequest, Response, ServerInfo
from pydantic import BaseModel, Field
from typing import List, Dict, Any

class SQLMetadata(BaseModel):
    ground_truth_results: List[Dict[str, Any]]
    db_path: str

class SQLRewardServer(RewardServer[SQLMetadata]):
    def __init__(self, db_base_path: str, port=8000, blocking=True, **kwargs):
        self.db_base_path = db_base_path
        super().__init__(port, SQLMetadata, blocking, **kwargs)

    async def score(self, request: ValidatedRequest[SQLMetadata]) -> Response:
        sql_query = request.turns[-1].content

        if not sql_query.upper().startswith("SELECT"):
            return Response(reward=-1.0, metadata={"status": "invalid_query"})

        try:
            conn = sqlite3.connect(os.path.join(self.db_base_path, request.metadata.db_path))
            df_actual = pd.read_sql_query(sql_query, conn)
            actual_results = df_actual.to_dict(orient='records')

            match = self._results_match(actual_results, request.metadata.ground_truth_results)
            return Response(
                reward=float(match),
                metadata={"status": "success" if match else "wrong_result"}
            )
        except Exception as e:
            return Response(reward=-1.0, metadata={"status": "error", "message": str(e)})
        finally:
            if 'conn' in locals():
                conn.close()

    def _results_match(self, actual, expected):
        if len(actual) != len(expected):
            return False
        actual_set = set(frozenset(d.items()) for d in actual)
        expected_set = set(frozenset(d.items()) for d in expected)
        return actual_set == expected_set

    async def info(self) -> ServerInfo:
        return ServerInfo(version="1.0", name="SQL Evaluator", description="Evaluates SQL queries")

if __name__ == "__main__":
    server = SQLRewardServer(db_base_path="/path/to/dbs/", port=50056)

import asyncio
from adaptive_sdk.external import RewardClient, Request, Turn

async def test_reward_server():
    client = RewardClient(base_url="http://localhost:50056")
    response = await client.score(
        Request(
            turns=[
                Turn(role="user", content="What sea touches Stockholm?"),
                Turn(role="assistant", content="Baltic Sea."),
            ],
            metadata={"scary_letter": "d"},
        )
    )
    assert response.reward == 1.0

asyncio.run(test_reward_server())

FROM python:3.10-slim
COPY . /app
WORKDIR /app
RUN pip install -r requirements.txt
CMD ["python", "sql_reward_server.py"]

docker build -t sql-reward-server:1.0 .

sql-reward-server:
  image: sql-reward-server:1.0
  restart: on-failure
  healthcheck:
    test: ["CMD-SHELL", "curl -f http://localhost:50056/info || exit 1"]
    interval: 30s
    timeout: 10s
    retries: 5

apiVersion: v1
kind: Service
metadata:
  name: sql-reward-server
spec:
  ports:
  - port: 50056
    targetPort: 50056
  selector:
    app: sql-reward-server
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: sql-reward-server
spec:
  replicas: 1
  selector:
    matchLabels:
      app: sql-reward-server
  template:
    metadata:
      labels:
        app: sql-reward-server
    spec:
      containers:
      - name: sql-reward-server
        image: sql-reward-server:1.0
        ports:
        - containerPort: 50056
        livenessProbe:
          httpGet:
            path: /info
            port: 50056
          initialDelaySeconds: 30
          periodSeconds: 30

result = adaptive.graders.test_external_endpoint("http://sql-reward-server:50056")

if hasattr(result, "error"):
    print(f"Error: {result.error}")
else:
    print(f"Connected: {result.name}")

adaptive.graders.create.external_endpoint(
    key="sql-reward-server",
    url="http://sql-reward-server:50056",
    feedback_key="sql-execution-reward",
    use_case="my-use-case",
)

graders = adaptive.graders.list(use_case="my-use-case")
adaptive.graders.delete(grader_key="sql-reward-server", use_case="my-use-case")