Calling LLMs

The AiSdk class provides a standardized way to call LLM services hosted in the AI SDK microservice. It handles HTTP requests, response parsing, and type conversion automatically.

Importing `AiSdk`

from services.ai.client import AiSdk

Basic Usage

AiSdk.call is a static async method that sends a POST request to the AI SDK service and returns a typed response.

result = await AiSdk.call(
    route: str,
    payload: BaseModel,
    response_type: type[T],
    timeout: float = 60,
    **kwargs: Any,
) -> T

Parameters

route: The API route path (e.g., "image/qa", "outline-assistant/server")
payload: A Pydantic BaseModel instance containing the request data
response_type: The expected response type (BaseModel subclass, str, or list)
timeout: Request timeout in seconds (default: 60)
**kwargs: Additional arguments passed to the HTTP client

Request Models

Request models must:

Extend BaseModel from Pydantic
Use exact field names that match the AI SDK API

class ImageQARequest(BaseModel):
    imageUrl: str  # Match AI SDK field names exactly
    text: str

Response Types

AiSdk.call supports three types of responses:

Pydantic Models

For structured responses, use a Pydantic model as the response_type:

from pydantic import BaseModel
from services.ai.client import AiSdk


class ImageQARequest(BaseModel):
    imageUrl: str
    text: str


class ImageQAResponse(BaseModel):
    passes: bool


async def ai_image_qa(request: ImageQARequest) -> ImageQAResponse:
    return await AiSdk.call("image/qa", request, ImageQAResponse, timeout=20)

String Responses

For plain text responses, use str as the response_type:

from pydantic import BaseModel
from services.ai.client import AiSdk


class SummarizeMessagesRequest(BaseModel):
    messages: list[dict[str, Any]]


async def ai_summarize_messages(request: SummarizeMessagesRequest) -> str:
    return await AiSdk.call("summarize-messages", request, str, timeout=60)

List Responses

For array responses, use list as the response_type. You’ll need to manually parse the list items since AiSdk.call returns a generic list.

from pydantic import BaseModel
from services.ai.client import AiSdk
from presentation.generation.models import OutlineChunk


class OutlineGenerationRequest(BaseModel):
    topic: str
    language: str
    length: OutlineLengthRequest


async def ai_generate_outline(
    request: OutlineGenerationRequest,
) -> list[OutlineChunk]:
    data = await AiSdk.call(
        "outline-assistant/server",
        request,
        list,
        timeout=120,
    )
    return [OutlineChunk(**outline_chunk) for outline_chunk in data]

Naming Convention

Functions that call AiSdk must be prefixed with ai_:

async def ai_image_qa(request: ImageQARequest) -> ImageQAResponse:
    return await AiSdk.call("image/qa", request, ImageQAResponse, timeout=20)

Timeout Configuration

Always set an appropriate timeout based on the expected operation duration. This allows for proper retry handling in Temporal.

Quick operations (image QA, keyword extraction): timeout=20
Standard operations (content generation): timeout=30-60
Long operations (outline generation, translation): timeout=120

# Quick operation
result = await AiSdk.call("image/qa", request, ImageQAResponse, timeout=20)

# Long operation
result = await AiSdk.call(
    "outline-assistant/server",
    request,
    list,
    timeout=120,
)