ag_kit_py.providers

The Model Provider system provides a unified interface for interacting with different Large Language Model (LLM) services. It abstracts away provider-specific implementations and offers a consistent API for creating chat completions, streaming responses, and managing tools across multiple LLM providers.

Installation

pip install ag_kit_py

Key Features

Unified Interface: Consistent API across multiple LLM providers
Multiple Providers: Support for OpenAI, Zhipu, Qwen, DeepSeek, Anthropic, and custom providers
Streaming Support: Real-time streaming responses for better user experience
Tool Calling: Function/tool calling capabilities with standardized format
LangChain Integration: Seamless integration with LangChain and LangGraph
Error Handling: Comprehensive error handling with retry mechanisms
Configuration Management: Flexible configuration with environment variable support

Supported Providers

Provider	Type	Status	Description
OpenAI	`openai`	✅ Available	OpenAI API and compatible endpoints
Zhipu	`zhipu`	🚧 Coming Soon	Zhipu AI (智谱AI)
Qwen	`qwen`	🚧 Coming Soon	Alibaba Qwen (通义千问)
DeepSeek	`deepseek`	🚧 Coming Soon	DeepSeek AI
Anthropic	`anthropic`	🚧 Coming Soon	Anthropic Claude
Custom	`custom`	✅ Available	Custom provider implementation

Quick Start

Using Factory Function (Recommended)

The simplest way to create a provider is using the factory function with environment variables:

import os
from ag_kit_py.providers import create_provider

# Set environment variables
os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["OPENAI_MODEL"] = "gpt-4o-mini"

# Create provider
provider = create_provider(provider_type="openai")

# Get LangChain model for use with LangGraph
chat_model = provider.get_langchain_model()

Direct Provider Creation

For more control, create providers directly:

from ag_kit_py.providers import OpenAIProvider, ModelProviderConfig

config = ModelProviderConfig(
    api_key="sk-...",
    default_model="gpt-4o-mini",
    temperature=0.7,
    timeout=30000
)

provider = OpenAIProvider(config)

Using with LangGraph Agents

from ag_kit_py.providers import create_provider
from ag_kit_py.agents import LangGraphAgent
from langgraph.graph import StateGraph, MessagesState

# Create provider
provider = create_provider(provider_type="openai")

# Get LangChain model
model = provider.get_langchain_model()

# Build LangGraph workflow
workflow = StateGraph(MessagesState)
workflow.add_node("chat", lambda state: {"messages": [model.invoke(state["messages"])]})
workflow.set_entry_point("chat")
workflow.set_finish_point("chat")

# Compile and create agent
graph = workflow.compile()
agent = LangGraphAgent(
    name="ChatAgent",
    description="A conversational agent",
    graph=graph
)

Core Concepts

Provider Types

Providers are identified by their type string:

from ag_kit_py.providers import ProviderType

# Available types
ProviderType.OPENAI      # "openai"
ProviderType.ZHIPU       # "zhipu"
ProviderType.QWEN        # "qwen"
ProviderType.DEEPSEEK    # "deepseek"
ProviderType.ANTHROPIC   # "anthropic"
ProviderType.CUSTOM      # "custom"

Configuration

All providers use the ModelProviderConfig dataclass:

from ag_kit_py.providers import ModelProviderConfig

config = ModelProviderConfig(
    api_key="your-api-key",           # Required: API key for authentication
    base_url=None,                     # Optional: Custom API endpoint
    default_model="gpt-4o-mini",       # Optional: Default model name
    temperature=0.7,                   # Optional: Sampling temperature (0.0-2.0)
    timeout=30000,                     # Optional: Request timeout in milliseconds
    max_retries=3,                     # Optional: Maximum retry attempts
    retry_delay=1000,                  # Optional: Delay between retries in ms
    proxy=None,                        # Optional: Proxy URL
    organization=None,                 # Optional: Organization ID
    project=None                       # Optional: Project ID
)

Chat Completion

Providers support both synchronous and streaming chat completions:

from ag_kit_py.providers import ChatCompletionParams

# Prepare parameters
params = ChatCompletionParams(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"}
    ],
    temperature=0.7,
    max_tokens=1000
)

# Non-streaming completion
response = await provider.create_completion(params)
print(response.choices[0]["message"]["content"])

# Streaming completion
async for chunk in provider.create_stream(params):
    content = chunk.choices[0]["delta"].get("content", "")
    print(content, end="", flush=True)

Tool/Function Calling

Providers support tool calling with standardized format:

from ag_kit_py.providers import ChatCompletionParams

# Define tools
tools = [
    {
        "name": "get_weather",
        "description": "Get current weather for a location",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "City name"
                }
            },
            "required": ["location"]
        }
    }
]

# Create completion with tools
params = ChatCompletionParams(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "What's the weather in Paris?"}],
    tools=tools,
    tool_choice="auto"
)

response = await provider.create_completion(params)

# Parse tool calls
tool_calls = provider.parse_tool_calls(response)
for call in tool_calls:
    print(f"Tool: {call.function['name']}")
    print(f"Arguments: {call.function['arguments']}")

Environment Variables

Providers support configuration through environment variables:

OpenAI Provider

# Required
OPENAI_API_KEY=sk-...

# Optional
OPENAI_MODEL=gpt-4o-mini
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_TEMPERATURE=0.7

Using Environment Variables

from ag_kit_py.providers import create_provider_from_env

# Automatically loads from environment
provider = create_provider_from_env("openai")

Error Handling

All providers use the ModelProviderError exception class:

from ag_kit_py.providers import ModelProviderError

try:
    response = await provider.create_completion(params)
except ModelProviderError as e:
    print(f"Error type: {e.error_type}")
    print(f"Message: {e.message}")
    print(f"Status: {e.status}")
    print(f"Code: {e.code}")

Error Types

authentication: Invalid API key or credentials
rate_limit: Rate limit exceeded
quota_exceeded: API quota exceeded
invalid_request: Invalid request parameters
server_error: Provider server error
timeout: Request timeout
unknown: Unknown error

Best Practices

1. Use Environment Variables

Store sensitive credentials in environment variables:

import os
from ag_kit_py.providers import create_provider

# Good: Use environment variables
os.environ["OPENAI_API_KEY"] = "sk-..."
provider = create_provider("openai")

# Bad: Hardcode credentials
provider = create_provider("openai", api_key="sk-...")  # Don't commit this!

2. Reuse Provider Instances

Create provider instances once and reuse them:

# Good: Create once, reuse
provider = create_provider("openai")
model = provider.get_langchain_model()

# Use model multiple times
response1 = await model.ainvoke(messages1)
response2 = await model.ainvoke(messages2)

3. Handle Errors Gracefully

Implement proper error handling with retries:

from ag_kit_py.providers import ModelProviderError
import asyncio

async def call_with_retry(provider, params, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await provider.create_completion(params)
        except ModelProviderError as e:
            if e.error_type in ["authentication", "invalid_request"]:
                # Don't retry non-retryable errors
                raise
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

4. Configure Timeouts

Set appropriate timeouts for your use case:

from ag_kit_py.providers import ModelProviderConfig

# For quick responses
config = ModelProviderConfig(
    api_key="sk-...",
    timeout=10000  # 10 seconds
)

# For long-running tasks
config = ModelProviderConfig(
    api_key="sk-...",
    timeout=120000  # 2 minutes
)

5. Use Streaming for Better UX

Use streaming for real-time responses:

async def stream_response(provider, params):
    """Stream response with proper error handling."""
    try:
        async for chunk in provider.create_stream(params):
            content = chunk.choices[0]["delta"].get("content", "")
            if content:
                yield content
    except ModelProviderError as e:
        yield f"\n[Error: {e.message}]"

Architecture

Class Hierarchy

BaseModelProvider (Abstract)
├── OpenAIProvider
├── ZhipuProvider (Coming Soon)
├── QwenProvider (Coming Soon)
├── DeepSeekProvider (Coming Soon)
└── AnthropicProvider (Coming Soon)

Key Methods

All providers implement these abstract methods:

create_completion(): Create a chat completion
create_stream(): Create a streaming chat completion
get_provider_name(): Get provider name
get_default_model(): Get default model name
format_tools(): Format tools for the provider
parse_tool_calls(): Parse tool calls from response
get_langchain_model(): Get LangChain-compatible model

Base Provider API - Abstract base class and interfaces
OpenAI Provider - OpenAI provider implementation
Factory Functions - Provider creation utilities
Agents Overview - Using providers with agents
LangGraph Agent - LangGraph integration

Next Steps

Learn about Base Provider API for custom implementations
Explore OpenAI Provider for detailed usage
Check Factory Functions for simplified creation
See LangGraph Agent for agent integration

Documentation Index

​ag_kit_py.providers

​Installation

​Key Features

​Supported Providers

​Quick Start

​Using Factory Function (Recommended)

​Direct Provider Creation

​Using with LangGraph Agents

​Core Concepts

​Provider Types

​Configuration

​Chat Completion

​Tool/Function Calling

​Environment Variables

​OpenAI Provider

​Using Environment Variables

​Error Handling

​Error Types

​Best Practices

​1. Use Environment Variables

​2. Reuse Provider Instances

​3. Handle Errors Gracefully

​4. Configure Timeouts

​5. Use Streaming for Better UX

​Architecture

​Class Hierarchy

​Key Methods

​Related Documentation

​Next Steps