vv-llm 0.3.110

Universal LLM interfaces for multi-provider chat and utilities

vv-llm

中文文档

Universal LLM interface layer for Python. One API, 16 backends, sync & async.

pip install vv-llm

Supported Backends

OpenAI | Anthropic | DeepSeek | Gemini | Qwen | Groq | Mistral | Moonshot | MiniMax | Yi | ZhiPuAI | Baichuan | StepFun | xAI | Ernie | Local

Also supports Azure OpenAI, Vertex AI, and AWS Bedrock deployments.

Quick Start

Configure

from vv_llm.settings import settings

settings.load({
    "VERSION": "2",
    "endpoints": [
        {
            "id": "openai-default",
            "api_base": "https://api.openai.com/v1",
            "api_key": "sk-...",
        }
    ],
    "backends": {
        "openai": {
            "models": {
                "gpt-4o": {
                    "id": "gpt-4o",
                    "endpoints": ["openai-default"],
                }
            }
        }
    }
})

Sync

from vv_llm.chat_clients import create_chat_client, BackendType

client = create_chat_client(BackendType.OpenAI, model="gpt-4o")
resp = client.create_completion([
    {"role": "user", "content": "Explain RAG in one sentence"}
])
print(resp.content)

Pass thinking explicitly when a provider supports Anthropic-style thinking control; omit it to keep the provider default:

resp = client.create_completion(
    messages=[{"role": "user", "content": "Answer directly"}],
    thinking={"type": "disabled"},
)

The legacy keyword API remains supported. New code can use a normalized request and typed thinking controls:

from vv_llm import ChatRequest, ChatRequestOptions, ThinkingPreference

resp = client.create(
    ChatRequest(
        messages=[{"role": "user", "content": "Answer directly"}],
        options=ChatRequestOptions(
            thinking=ThinkingPreference.disabled(),
            max_tokens=512,
        ),
    )
)

print(client.capabilities.thinking)

Use ThinkingPreference.default() to preserve the provider default, enabled() or enabled(budget_tokens=...) to opt in, and disabled() to opt out explicitly.

Middleware, Retry, And Metadata

The legacy client remains unchanged. Wrap it only when the application needs a versioned middleware chain, classified retry, or execution metadata:

from vv_llm import ChatMiddlewareV1, ChatRequest, MiddlewareChatClient, RetryPolicy

class TraceMiddleware(ChatMiddlewareV1):
    def on_request(self, context, request):
        context.attributes["trace_id"] = "request-42"
        return request

runtime = MiddlewareChatClient(
    client,
    [TraceMiddleware()],
    retry_policy=RetryPolicy(max_attempts=3, total_timeout=20),
)
result = runtime.create_with_metadata(
    ChatRequest(messages=[{"role": "user", "content": "Answer directly"}])
)

print(result.response.content)
print(result.metadata.provider, result.metadata.attempts, result.metadata.latency_ms)

ErrorKind distinguishes authentication, rate limiting, network, timeout, invalid request, context length, content policy, missing model, provider internal, serialization, and configuration failures. The default retry policy retries only transient kinds and respects Retry-After, exponential backoff, jitter, and an optional total deadline.

Explicit Registry And Fallback

Fallback is opt-in and ordered. Every registration declares model capabilities, so an incompatible route is skipped without sending a request:

from vv_llm import FallbackChatClient, FallbackRoute, ProviderRegistry

registry = ProviderRegistry()
registry.register(
    "primary",
    lambda: primary_client,
    capabilities=primary_client.capabilities,
)
registry.register(
    "secondary",
    lambda: secondary_client,
    capabilities=secondary_client.capabilities,
)
runtime = FallbackChatClient(
    registry,
    [
        FallbackRoute("primary", "primary-model"),
        FallbackRoute("secondary", "secondary-model"),
    ],
)

Authentication and invalid-request errors do not fall back by default. Streaming may switch routes only while establishing the stream or before its first visible chunk; after output begins, later errors are returned without replay.

Streaming

for chunk in client.create_stream([
    {"role": "user", "content": "Write a haiku"}
]):
    if chunk.content:
        print(chunk.content, end="")

Async

import asyncio
from vv_llm.chat_clients import create_async_chat_client, BackendType

async def main():
    client = create_async_chat_client(BackendType.OpenAI, model="gpt-4o")
    resp = await client.create_completion([
        {"role": "user", "content": "hello"}
    ])
    print(resp.content)

asyncio.run(main())

Embedding & Rerank

from vv_llm.settings import settings

settings.load({
    "VERSION": "2",
    "endpoints": [
        {
            "id": "siliconflow",
            "api_base": "https://api.siliconflow.cn/v1",
            "api_key": "sk-...",
        }
    ],
    "backends": {},
    "embedding_backends": {
        "siliconflow": {
            "models": {
                "BAAI/bge-large-zh-v1.5": {
                    "id": "BAAI/bge-large-zh-v1.5",
                    "endpoints": ["siliconflow"],
                    "protocol": "openai_embeddings",
                }
            }
        }
    },
    "rerank_backends": {
        "siliconflow": {
            "models": {
                "BAAI/bge-reranker-v2-m3": {
                    "id": "BAAI/bge-reranker-v2-m3",
                    "endpoints": ["siliconflow"],
                    "protocol": "custom_json_http",
                    "request_mapping": {
                        "method": "POST",
                        "path": "/rerank",
                        "body_template": {
                            "model": "${model_id}",
                            "query": "${query}",
                            "documents": "${documents}",
                        },
                    },
                    "response_mapping": {
                        "results_path": "$.results[*]",
                        "field_map": {
                            "index": "$.index",
                            "relevance_score": "$.relevance_score",
                        },
                    },
                }
            }
        }
    },
})

from vv_llm.embedding_clients import create_embedding_client
from vv_llm.rerank_clients import create_rerank_client

embedding_client = create_embedding_client("siliconflow", model="BAAI/bge-large-zh-v1.5")
embedding_resp = embedding_client.create_embeddings(input="hello world")
print(len(embedding_resp.data[0].embedding))

rerank_client = create_rerank_client("siliconflow", model="BAAI/bge-reranker-v2-m3")
rerank_resp = rerank_client.rerank(
    query="Apple",
    documents=["apple", "banana", "fruit", "vegetable"],
)
print(rerank_resp.results[0].index, rerank_resp.results[0].relevance_score)

import asyncio
from vv_llm.embedding_clients import create_async_embedding_client
from vv_llm.rerank_clients import create_async_rerank_client

async def main():
    embedding_client = create_async_embedding_client("siliconflow", model="BAAI/bge-large-zh-v1.5")
    rerank_client = create_async_rerank_client("siliconflow", model="BAAI/bge-reranker-v2-m3")

    emb = await embedding_client.create_embeddings(input=["a", "b"])
    rr = await rerank_client.rerank(query="Apple", documents=["apple", "banana"])
    print(len(emb.data), len(rr.results))

asyncio.run(main())

Features

• Unified interface — same create_completion / create_stream API across all providers
• Embedding & rerank — unified sync/async retrieval clients with normalized outputs
• Type-safe factory — create_chat_client(BackendType.X) returns the correct client type
• Multi-endpoint — configure multiple endpoints per backend with random selection and failover
• Tool calling — normalized tool/function calling across providers
• Multimodal — text + image inputs where supported
• Thinking/reasoning — access chain-of-thought from Claude, DeepSeek Reasoner, etc.
• Token counting — per-model tokenizers (tiktoken, deepseek-tokenizer, qwen-tokenizer)
• Rate limiting — RPM/TPM controls with memory, Redis, or DiskCache backends
• Context length control — automatic message truncation to fit model limits
• Prompt caching — Anthropic prompt caching support
• Retry with backoff — configurable retry logic for transient failures
• Versioned middleware — stable v1 request, response, and error hooks outside provider adapters
• Classified errors — provider-neutral error kinds with retryability and request context
• Explicit fallback — registered, ordered, capability-aware routes with no hidden provider switching
• Scripted testing — deterministic completion/error/stream scripts for conformance tests

Examples

Runnable typed-thinking, sync/async streaming, middleware metadata, and explicit fallback examples are available in examples/.

Cache Usage Semantics

OpenAI-compatible chat completions report cache reads through usage.prompt_tokens_details.cached_tokens. usage.prompt_tokens remains the total input token count, so consumers can calculate uncached input as prompt_tokens - cached_tokens. This path intentionally does not populate Anthropic's cache_read_input_tokens field because Anthropic defines its base input_tokens as uncached input.

For generic OpenAI-compatible backends, omitted cache-read fields remain unknown, while an explicit cached_tokens: 0 is preserved as an observed zero. Moonshot may omit both top-level cached_tokens and prompt_tokens_details on a cold request; only in that fully omitted case does vv-llm project prompt_tokens_details.cached_tokens = 0 from the provider contract. Explicit null or invalid cache values remain unknown.

Utilities

from vv_llm.chat_clients import format_messages, get_token_counts, get_message_token_counts

Function	Description
format_messages	Normalize multimodal/tool messages across formats
get_token_counts	Count tokens for a text string
get_message_token_counts	Count tokens for a message list

Optional Dependencies

pip install 'vv-llm[redis]'      # Redis rate limiting
pip install 'vv-llm[diskcache]'  # DiskCache rate limiting
pip install 'vv-llm[server]'     # FastAPI token server
pip install 'vv-llm[vertex]'     # Google Vertex AI
pip install 'vv-llm[bedrock]'    # AWS Bedrock

Project Structure

src/vv_llm/
  chat_clients/    # Per-backend clients + factory
  embedding_clients/  # Embedding clients + factory
  rerank_clients/     # Rerank clients + factory
  retrieval_clients/  # Shared retrieval client internals
  settings/        # Configuration management
  types/           # Type definitions & enums
  utilities/       # Rate limiting, retry, media processing, token counting
  server/          # Optional token counting server

tests/unit/        # Unit tests
tests/live/        # Live integration tests (requires real API keys)

Development

pdm install -d          # Install dev dependencies
pdm run lint            # Ruff linter
pdm run format-check    # Ruff format check
pdm run type-check      # Ty type checker
pdm run test            # Unit tests
pdm run test-live       # Live tests (needs real endpoints)

License

MIT