LLM 接口层

from abc import ABC, abstractmethod
from typing import Iterable, Dict, Optional, Literal, AsyncGenerator
from openai.types.chat.chat_completion import ChatCompletion
from openai.types.chat.chat_completion_chunk import ChatCompletionChunk

class LLM_Interface(ABC):
    @abstractmethod
    async def chat(
        self,
        trace_id: str = get_current_trace_id(),
        stream: Literal[False] = False,
        messages: Iterable[Dict[str, str]] = [{"role": "user", "content": ""}],
        timeout: Optional[int] = None,
        *args,
        **kwargs,
    ) -> ChatCompletion:
        pass

    @abstractmethod
    async def chat_stream(
        self,
        trace_id: str = get_current_trace_id(),
        stream: Literal[True] = True,
        messages: Iterable[Dict[str, str]] = [{"role": "user", "content": ""}],
        timeout: Optional[int] = None,
        *args,
        **kwargs,
    ) -> AsyncGenerator[ChatCompletionChunk, None]:
        if False:
            yield ChatCompletionChunk(id="", created=0, model="", object="", choices=[])

堆结构：[(任务数, API密钥), ...]
堆顶永远是任务数最少的密钥

from SimpleLLMFunc import OpenAICompatible, llm_function

models = OpenAICompatible.load_from_json_file("provider.json")

fast_llm = models["openai"]["gpt-3.5-turbo"]
powerful_llm = models["openai"]["gpt-4"]
deepseek_llm = models["deepseek"]["deepseek-chat"]

@llm_function(llm_interface=fast_llm)
async def simple_task(text: str) -> str:
    """简单任务使用快速模型"""
    pass

@llm_function(llm_interface=powerful_llm)
async def complex_task(text: str) -> str:
    """复杂任务使用强大模型"""
    pass

@llm_function(llm_interface=llm)
async def robust_call(text: str) -> str:
    """可靠的 LLM 调用"""
    pass

async def call_with_fallback():
    try:
        return await robust_call("test")
    except Exception as e:
        print(f"主模型失败: {e}")
        return await backup_call("test")

from SimpleLLMFunc.hooks.events import LLMCallEndEvent
from SimpleLLMFunc.hooks.stream import is_event_yield

async for output in summarize_text("..."):
    if is_event_yield(output) and isinstance(output.event, LLMCallEndEvent):
        print(output.event.usage)

least_loaded = llm.key_pool.get_least_loaded_key()
print(f"最低负载密钥: {least_loaded}")
print(llm.get_rate_limit_status())

import asyncio
from typing import Optional

async def call_with_exponential_backoff(
    llm_call,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> Optional[str]:
    for attempt in range(max_retries):
        try:
            return await llm_call()
        except Exception:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(base_delay * (2 ** attempt))

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("SimpleLLMFunc")
logger.setLevel(logging.DEBUG)