smartllm 0.1.16

A unified async Python wrapper for multiple LLM providers with OpenAI Response API and reasoning support

SmartLLM

A unified async Python wrapper for multiple LLM providers with a consistent interface.

Features

• Unified Interface — Single API for OpenAI and AWS Bedrock
• Async/Await — Built on asyncio for concurrent requests
• Smart Caching — Two-level cache (local JSON + optional DynamoDB)
• Auto Retry — Exponential backoff for transient failures
• Structured Output — Native Pydantic model support
• Streaming — Real-time streaming responses
• Rate Limiting — Built-in concurrency control per model
• Reasoning Models — Full support including reasoning_effort and reasoning_tokens
• Extended Thinking (Bedrock) — Claude extended thinking with two-pass structured output
• Progress Callbacks — Optional on_progress for real-time events

Installation

pip install smartllm[openai]   # OpenAI only
pip install smartllm[bedrock]  # AWS Bedrock only
pip install smartllm[all]      # All providers

Quick Start

import asyncio
from smartllm import LLMClient, TextRequest

async def main():
    async with LLMClient(provider="openai") as client:
        response = await client.generate_text(
            TextRequest(prompt="What is the capital of France?")
        )
        print(response.text)

asyncio.run(main())

Configuration

Environment Variables

OpenAI:

export OPENAI_API_KEY="your-api-key"
export OPENAI_MODEL="gpt-4o-mini"  # optional

AWS Bedrock:

export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_REGION="us-east-1"
export BEDROCK_MODEL="anthropic.claude-3-sonnet-20240229-v1:0"  # optional

Explicit credentials are optional. If omitted, boto3's default credential chain is used — including EC2 instance profiles, ECS task roles, Lambda execution roles, and ~/.aws/credentials.

Programmatic Configuration

from smartllm import LLMClient, LLMConfig

config = LLMConfig(
    provider="openai",
    api_key="your-api-key",
    default_model="gpt-4o",
    temperature=0.7,
    max_tokens=2048,
    max_retries=3,
)

async with LLMClient(config) as client:
    ...

Usage Examples

Multi-turn Conversations

from smartllm import LLMClient, MessageRequest, Message

async with LLMClient(provider="openai") as client:
    messages = [
        Message(role="user", content="My name is Alice."),
        Message(role="assistant", content="Nice to meet you, Alice!"),
        Message(role="user", content="What's my name?"),
    ]
    response = await client.send_message(MessageRequest(messages=messages))
    print(response.text)  # "Your name is Alice."

Structured Output

from pydantic import BaseModel
from smartllm import LLMClient, TextRequest

class Person(BaseModel):
    name: str
    age: int

async with LLMClient(provider="openai") as client:
    response = await client.generate_text(
        TextRequest(prompt="Return a person named John, age 30.", response_format=Person)
    )
    print(response.structured_data.name)  # "John"

Streaming

async with LLMClient(provider="openai") as client:
    async for chunk in client.generate_text_stream(
        TextRequest(prompt="Write a short poem.", stream=True)
    ):
        print(chunk.text, end="", flush=True)

Reasoning Models

response = await client.generate_text(
    TextRequest(
        prompt="Solve: what is the 100th Fibonacci number?",
        reasoning_effort="high",  # "low", "medium", or "high"
    )
)
print(response.text)
print(f"Reasoning tokens: {response.reasoning_tokens}")

Note: reasoning models do not support temperature. Passing a value other than 1 raises ValueError.

Extended Thinking (Bedrock/Claude)

Claude models on Bedrock support extended thinking, giving the model a token budget to reason step-by-step before answering.

async with LLMClient(provider="bedrock") as client:
    # Using reasoning_effort (maps to budget: low=1024, medium=4096, high=16000)
    response = await client.generate_text(
        TextRequest(
            prompt="Analyze the tradeoffs of event sourcing vs CRUD.",
            model="eu.anthropic.claude-sonnet-4-6",
            reasoning_effort="high",
        )
    )
    print(response.text)
    print(f"Reasoning tokens: {response.reasoning_tokens}")
    print(f"Thinking: {response.metadata.get('thinking', '')[:200]}")

For precise control, use budget_tokens directly (overrides reasoning_effort):

response = await client.generate_text(
    TextRequest(
        prompt="Solve this step by step...",
        model="eu.anthropic.claude-sonnet-4-6",
        budget_tokens=8192,  # Explicit token budget (minimum 1024)
    )
)

Extended Thinking + Structured Output

When both reasoning_effort (or budget_tokens) and response_format are set, SmartLLM uses a two-pass approach:

1. Pass 1 — Sends the prompt with extended thinking enabled. Claude reasons through the problem and produces a text answer.
2. Pass 2 — Sends the text answer to a second call with forced tool use to extract it into the Pydantic model.

from pydantic import BaseModel
from typing import List

class Analysis(BaseModel):
    topic: str
    pros: List[str]
    cons: List[str]
    recommendation: str

response = await client.generate_text(
    TextRequest(
        prompt="Should we use microservices or a monolith?",
        model="eu.anthropic.claude-sonnet-4-6",
        reasoning_effort="medium",
        response_format=Analysis,
    )
)
print(response.structured_data.recommendation)
print(response.metadata["pass1_tokens"])  # {"input": ..., "output": ...}
print(response.metadata["pass2_tokens"])  # {"input": ..., "output": ...}

The two-pass approach is needed because Claude's extended thinking is incompatible with forced tool use (tool_choice: {"type": "tool"}). The result is cached as a single entry — on cache hit, both passes are skipped.

Streaming with Extended Thinking

When streaming with thinking enabled, thinking chunks are yielded with metadata={"type": "thinking"}:

async for chunk in client.generate_text_stream(
    TextRequest(prompt="Explain quantum entanglement.", reasoning_effort="medium", stream=True)
):
    if chunk.metadata.get("type") == "thinking":
        print(f"[thinking] {chunk.text}", end="")
    else:
        print(chunk.text, end="")

OpenAI API Types

# Responses API (default, recommended)
TextRequest(prompt="Hello", api_type="responses")

# Chat Completions API (legacy)
TextRequest(prompt="Hello", api_type="chat_completions")

Concurrent Requests

tasks = [client.generate_text(TextRequest(prompt=p)) for p in prompts]
responses = await asyncio.gather(*tasks)

Progress Callbacks

async def on_progress(event):
    print(event)

response = await client.generate_text(
    TextRequest(prompt="Hello", on_progress=on_progress)
)

Events: llm_started, llm_done, cache_hit (with cache_source, cache_key), error (with message). Each event dict includes event, ts, prompt, model, provider. llm_done and cache_hit also include input_tokens, output_tokens, reasoning_tokens, cached_tokens.

DynamoDB Caching

async with LLMClient(provider="openai", dynamo_table_name="my-llm-cache") as client:
    ...

Requires AWS credentials with DynamoDB access. Table is auto-created if it doesn't exist.

Provider-Specific Clients

from smartllm.openai import OpenAILLMClient, OpenAIConfig
from smartllm.bedrock import BedrockLLMClient, BedrockConfig

async with OpenAILLMClient(OpenAIConfig(api_key="...")) as client:
    models = await client.list_available_models()

async with BedrockLLMClient(BedrockConfig(aws_region="us-east-1")) as client:
    models = await client.list_available_model_ids()

API Reference

TextRequest Parameters

Parameter	Type	Description	Default
prompt	str	Input text prompt	Required
model	str	Model ID	Config default
temperature	float	Sampling temperature (0–1)	0
max_tokens	int	Maximum output tokens	2048
top_p	float	Nucleus sampling	1.0
system_prompt	str	System context	None
stream	bool	Enable streaming	False
response_format	BaseModel	Pydantic model for structured output	None
use_cache	bool	Enable caching	True
clear_cache	bool	Clear cache before request	False
api_type	str	"responses" or "chat_completions"	"responses"
reasoning_effort	str	"low", "medium", or "high"	None
budget_tokens	int	Explicit thinking budget in tokens (Bedrock/Claude). Overrides reasoning_effort mapping. Minimum 1024.	None
on_progress	Callable	Progress event callback (sync or async)	None

TextResponse Fields

Field	Type	Description
text	str	Generated text
model	str	Model that generated the response
stop_reason	str	Reason generation stopped
input_tokens	int	Input token count
output_tokens	int	Output token count
reasoning_tokens	int	Reasoning/thinking tokens used (OpenAI reasoning models and Bedrock extended thinking)
cached_tokens	int	Prompt cache tokens (OpenAI only, 0 otherwise)
timestamp	str | None	ISO 8601 UTC timestamp of the original API call
elapsed_seconds	float | None	Duration of the original API call in seconds
metadata	dict	Request context: prompt/messages and response_format JSON schema
structured_data	BaseModel | None	Parsed Pydantic object (when response_format was set)
cache_source	str	"miss", "l1" (local), or "l2" (DynamoDB)
cache_key	str | None	Cache key for this request

Structured Output Error Handling

When using response_format, two error conditions are raised explicitly:

Truncated output — if the provider cuts off the response before the structured output is complete, a ValueError is raised:

try:
    response = await client.generate_text(
        TextRequest(prompt="...", response_format=MyModel, max_tokens=100)
    )
except ValueError as e:
    print(e)  # "Bedrock truncated structured output (stop_reason=max_tokens)"
             # "OpenAI truncated structured output (finish_reason=length)"
             # "OpenAI truncated structured output (status=incomplete)"

Increase max_tokens to avoid this.

Provider serialization quirks — Bedrock occasionally returns list fields as JSON strings rather than inline arrays. Pydantic's model_validate is used internally to handle coercion where possible. If your model has list fields and you still see ValidationError, add a field validator:

import json
from pydantic import BaseModel, field_validator

class BookList(BaseModel):
    books: list[str]

    @field_validator("books", mode="before")
    @classmethod
    def parse_json_string(cls, v):
        if isinstance(v, str):
            return json.loads(v)
        return v

Caching

Responses are cached automatically when temperature=0, when using a reasoning model, or when extended thinking is enabled. Streaming responses are never cached.

Cache key is derived from: model, prompt (or messages), max_tokens, top_p, system_prompt, response_format, api_type, reasoning_effort, budget_tokens.

What is stored:

Field	Description
text	Raw response text
model	Model used
stop_reason	Stop reason
input_tokens	Input token count
output_tokens	Output token count
reasoning_tokens	Reasoning token count
cached_tokens	Prompt cache token count
timestamp	ISO 8601 UTC timestamp of the original API call
elapsed_seconds	Duration of the original API call in seconds
metadata.prompt	Original prompt (or messages) — stored in top-level cache metadata, not duplicated in data
metadata.response_format	JSON schema of requested output format
structured_data	Parsed Pydantic object (as dict)

timestamp and elapsed_seconds are stored and restored on cache hits — they reflect when the original API call was made and how long it took.

response1 = await client.generate_text(TextRequest(prompt="What is 2+2?", temperature=0))
print(response1.cache_source)  # "miss"

response2 = await client.generate_text(TextRequest(prompt="What is 2+2?", temperature=0))
print(response2.cache_source)  # "l1" or "l2"

# Force refresh
response3 = await client.generate_text(TextRequest(prompt="What is 2+2?", temperature=0, clear_cache=True))

Development

git clone https://github.com/Redundando/smartllm.git
cd smartllm
pip install -e .[all,dev]

pytest tests/unit/ -v
pytest tests/integration/ --model gpt-4o

License

MIT — see LICENSE.
Issues: GitHub Issues