"""LLM HTTP Client — handles actual API calls. Supports both OpenAI SDK and raw httpx for flexibility. Provides both streaming and non-streaming call modes. """ from __future__ import annotations import logging from typing import Any, Callable, Optional, cast from butler.transport import get_transport from butler.transport.anthropic_transport import _STOP_REASON_MAP from butler.transport.interruptible_client import run_interruptible from butler.transport.llm_client_ops import ( iter_stream_safe, merge_thinking_headers_safe, wire_tools_or_empty_loud, ) from butler.transport.streaming_signal import notify_complete_tool_calls_from_stream from butler.transport.think_scrubber import StreamingThinkScrubber from butler.transport.types import NormalizedResponse, ToolCall, Usage from butler.transport.providers import ProviderProfile, get_provider logger = logging.getLogger(__name__) class LLMClient: """Stateful LLM API client for a specific provider configuration.""" def __init__( self, *, provider: str = "", model: str = "", api_key: Optional[str] = None, base_url: Optional[str] = None, api_mode: Optional[str] = None, max_tokens: Optional[int] = None, temperature: Optional[float] = None, timeout: int = 120, ): self.provider_name = provider self.model = model self._api_key = api_key self._base_url = base_url self._api_mode = api_mode self.max_tokens = max_tokens self.temperature = temperature self.timeout = timeout self._profile: Optional[ProviderProfile] = None self._client: Any = None # R2-10: Stash the original exception when tool wiring fails so the # caller can inspect it. ``None`` means the most recent call wired # tools successfully (or no tools were provided). self._last_tool_wire_error: Optional[BaseException] = None self._resolve_config() def _resolve_config(self) -> None: if self.provider_name: self._profile = get_provider(self.provider_name) if self._profile: if not self._base_url: self._base_url = self._profile.base_url if not self._api_key: self._api_key = self._profile.resolve_api_key() if not self._api_mode: self._api_mode = self._profile.api_mode if not self.model: self.model = self._profile.default_model if self.max_tokens is None: self.max_tokens = self._profile.default_max_tokens if not self._api_mode: self._api_mode = self._detect_api_mode() def _detect_api_mode(self) -> str: url = (self._base_url or "").lower().rstrip("/") if url.endswith("/anthropic"): return "anthropic_messages" return "chat_completions" @property def api_mode(self) -> str: return self._api_mode or "chat_completions" _MAX_RETRIES = 0 # 重试由外层 llm_retry 中间件统一处理，关闭 SDK 默认重试避免双层放大（Sprint 10 PERF-NEW-3） def _wire_tools_or_empty(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]: """R2-10: Wire tools for the active provider; on failure, return []. Behaviour change vs. the previous bad_fallback: when ``wire_tools_for_provider`` raises, we used to fall back to ``transport.convert_tools(tools)`` (generic OpenAI-style schema). Providers with strict schema validation (Anthropic, certain OpenAI-compatible gateways) would then reject the request with a 400 *after* the model had already emitted tool_calls based on the "known" tool set, leaving the caller with a half-broken turn. The safe fallback is an empty tool list: the model proceeds without any tools, so it cannot emit tool_calls the provider would reject. The original exception is preserved on ``self._last_tool_wire_error`` for caller inspection and is logged at ERROR with a full traceback. """ wired, exc = wire_tools_or_empty_loud( self.provider_name or "", tools, api_mode=self.api_mode, ) self._last_tool_wire_error = exc return cast(list[dict[str, Any]], wired) def _get_openai_client(self) -> Any: if self._client is None: try: from openai import OpenAI api_key = self._api_key if not api_key: logger.warning("OpenAI client created without API key; calls will likely fail") api_key = "dummy" self._client = OpenAI( api_key=api_key, base_url=self._base_url, timeout=self.timeout, max_retries=self._MAX_RETRIES, ) except ImportError: raise RuntimeError("openai package required: pip install openai") return self._client def _get_anthropic_client(self) -> Any: if self._client is None: api_key = self._api_key if not api_key: logger.warning("Anthropic client created without API key; calls will likely fail") api_key = "dummy" try: from anthropic import Anthropic self._client = Anthropic( api_key=api_key, base_url=self._base_url, timeout=self.timeout, max_retries=self._MAX_RETRIES, ) except ImportError: try: from openai import OpenAI self._client = OpenAI( api_key=api_key, base_url=self._base_url, timeout=self.timeout, max_retries=self._MAX_RETRIES, ) self._api_mode = "chat_completions" except ImportError: raise RuntimeError("anthropic or openai package required") return self._client def complete( self, messages: list[dict[str, Any]], tools: Optional[list[dict[str, Any]]] = None, check_interrupt: Optional[Callable[[], bool]] = None, stale_timeout: Optional[float] = None, **kwargs: Any, ) -> NormalizedResponse: """Non-streaming completion.""" transport = get_transport(self.api_mode) if transport is None: raise ValueError(f"No transport for api_mode={self.api_mode}") converted_tools = None if tools: converted_tools = self._wire_tools_or_empty(tools) params = { "temperature": kwargs.get("temperature", self.temperature), "max_tokens": kwargs.get("max_tokens", self.max_tokens), "timeout": kwargs.get("timeout", self.timeout), "stream": False, "provider": self.provider_name, "base_url": self._base_url, } params = {k: v for k, v in params.items() if v is not None} api_kwargs = transport.build_kwargs( model=self.model, messages=messages, tools=converted_tools, **params, ) stale = stale_timeout if stale_timeout is not None else min(float(self.timeout or 120), 90.0) def _do() -> NormalizedResponse: response = self._raw_call(api_kwargs) return transport.normalize_response(response) return cast( NormalizedResponse, run_interruptible( _do, check_interrupt=check_interrupt, stale_timeout=stale, ), ) def stream( self, messages: list[dict[str, Any]], tools: Optional[list[dict[str, Any]]] = None, on_delta: Optional[Callable[[str], None]] = None, on_tool_call_ready: Optional[Callable[[int, str, str, dict[str, Any]], None]] = None, check_interrupt: Optional[Callable[[], bool]] = None, stale_timeout: Optional[float] = None, **kwargs: Any, ) -> NormalizedResponse: """Streaming completion with delta callback.""" transport = get_transport(self.api_mode) if transport is None: raise ValueError(f"No transport for api_mode={self.api_mode}") converted_tools = None if tools: converted_tools = self._wire_tools_or_empty(tools) params = { "temperature": kwargs.get("temperature", self.temperature), "max_tokens": kwargs.get("max_tokens", self.max_tokens), "timeout": kwargs.get("timeout", self.timeout), "stream": True, "provider": self.provider_name, "base_url": self._base_url, } params = {k: v for k, v in params.items() if v is not None} api_kwargs = transport.build_kwargs( model=self.model, messages=messages, tools=converted_tools, **params, ) stale = stale_timeout if stale_timeout is not None else min(float(self.timeout or 120), 90.0) scrubber = StreamingThinkScrubber() def _wrapped_delta(delta: str) -> None: if on_delta: visible = scrubber.feed(delta) if visible: on_delta(visible) def _do() -> NormalizedResponse: try: return self._stream_call( api_kwargs, on_delta=_wrapped_delta if on_delta else None, on_tool_call_ready=on_tool_call_ready, transport=transport, ) finally: if on_delta: tail = scrubber.flush() if tail: on_delta(tail) return cast( NormalizedResponse, run_interruptible( _do, check_interrupt=check_interrupt, stale_timeout=stale, ), ) def _raw_call(self, api_kwargs: dict[str, Any]) -> Any: """Execute the actual API call.""" api_kwargs.pop("stream", None) if self.api_mode == "anthropic_messages": return self._call_anthropic(api_kwargs) return self._call_openai(api_kwargs) def _call_openai(self, api_kwargs: dict[str, Any]) -> Any: client = self._get_openai_client() return client.chat.completions.create(**api_kwargs) def _call_anthropic(self, api_kwargs: dict[str, Any]) -> Any: api_kwargs = merge_thinking_headers_safe( api_kwargs, provider=self.provider_name or "", model=self.model or "", ) client = self._get_anthropic_client() if hasattr(client, "messages"): return client.messages.create(**api_kwargs) return client.chat.completions.create(**api_kwargs) def _stream_call( self, api_kwargs: dict[str, Any], on_delta: Optional[Callable[[str], None]] = None, on_tool_call_ready: Optional[Callable[[int, str, str, dict[str, Any]], None]] = None, transport: Any = None, ) -> NormalizedResponse: """Execute streaming API call and collect into NormalizedResponse.""" api_kwargs["stream"] = True collected_content = [] collected_reasoning = [] collected_tool_calls: dict[int, dict[str, Any]] = {} finish_reason = "stop" usage = None if self.api_mode == "anthropic_messages": return self._stream_anthropic( api_kwargs, on_delta, transport, on_tool_call_ready=on_tool_call_ready, ) def _collect_openai_stream() -> str: nonlocal finish_reason, usage client = self._get_openai_client() stream = client.chat.completions.create(**api_kwargs) for chunk in stream: if not hasattr(chunk, "choices") or not chunk.choices: continue delta = chunk.choices[0].delta fr = chunk.choices[0].finish_reason if fr: finish_reason = fr text = getattr(delta, "content", None) if text: collected_content.append(text) if on_delta: on_delta(text) reasoning = getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) if isinstance(reasoning, str) and reasoning: collected_reasoning.append(reasoning) tcs = getattr(delta, "tool_calls", None) if tcs: for tc in tcs: idx = getattr(tc, "index", 0) if idx not in collected_tool_calls: collected_tool_calls[idx] = { "id": getattr(tc, "id", None), "name": "", "arguments": "", } entry = collected_tool_calls[idx] if getattr(tc, "id", None): entry["id"] = tc.id fn = getattr(tc, "function", None) if fn: if getattr(fn, "name", None): entry["name"] = fn.name if getattr(fn, "arguments", None): entry["arguments"] += fn.arguments notify_complete_tool_calls_from_stream( collected_tool_calls, on_tool_call_ready, ) raw_usage = getattr(chunk, "usage", None) if raw_usage: usage = Usage( prompt_tokens=getattr(raw_usage, "prompt_tokens", 0), completion_tokens=getattr(raw_usage, "completion_tokens", 0), total_tokens=getattr(raw_usage, "total_tokens", 0), ) return finish_reason finish_reason = iter_stream_safe( _collect_openai_stream, collected_content=collected_content, ) notify_complete_tool_calls_from_stream( collected_tool_calls, on_tool_call_ready, ) tool_calls = None if collected_tool_calls: tool_calls = [] for idx in sorted(collected_tool_calls.keys()): tc = collected_tool_calls[idx] tool_calls.append(ToolCall( id=tc["id"], name=tc["name"], arguments=tc["arguments"] or "{}", )) return NormalizedResponse( content="".join(collected_content) or None, tool_calls=tool_calls or None, finish_reason=finish_reason, reasoning="".join(collected_reasoning) or None, usage=usage, ) def _stream_anthropic( self, api_kwargs: dict[str, Any], on_delta: Optional[Callable[[str], None]], transport: Any, on_tool_call_ready: Optional[Callable[[int, str, str, dict[str, Any]], None]] = None, ) -> NormalizedResponse: api_kwargs.pop("stream", None) api_kwargs = merge_thinking_headers_safe( api_kwargs, provider=self.provider_name or "", model=self.model or "", label="llm_client.thinking_headers.stream", ) client = self._get_anthropic_client() if not hasattr(client, "messages"): api_kwargs["stream"] = True return self._stream_call( api_kwargs, on_delta=on_delta, transport=transport, on_tool_call_ready=on_tool_call_ready, ) collected_text: list[str] = [] tool_calls: list[ToolCall] = [] current_tool: Optional[dict[str, Any]] = None finish_reason = "stop" usage = None with client.messages.stream(**api_kwargs) as stream: for event in stream: etype = getattr(event, "type", "") if etype == "content_block_start": block = getattr(event, "content_block", None) if block and getattr(block, "type", "") == "tool_use": current_tool = { "id": getattr(block, "id", None), "name": getattr(block, "name", ""), "arguments": "", } elif etype == "content_block_delta": delta = getattr(event, "delta", None) if delta: dtype = getattr(delta, "type", "") if dtype == "text_delta": text = getattr(delta, "text", "") if text: collected_text.append(text) if on_delta: on_delta(text) elif dtype == "input_json_delta" and current_tool: partial = getattr(delta, "partial_json", "") if partial: current_tool["arguments"] += partial elif etype == "content_block_stop": if current_tool: notify_complete_tool_calls_from_stream( {len(tool_calls): current_tool}, on_tool_call_ready, ) tool_calls.append(ToolCall( id=current_tool["id"], name=current_tool["name"], arguments=current_tool["arguments"] or "{}", )) current_tool = None elif etype == "message_delta": delta = getattr(event, "delta", None) if delta: sr = getattr(delta, "stop_reason", None) if sr: finish_reason = _STOP_REASON_MAP.get(sr, "stop") raw_usage = getattr(event, "usage", None) if raw_usage: usage = Usage( prompt_tokens=getattr(raw_usage, "input_tokens", 0), completion_tokens=getattr(raw_usage, "output_tokens", 0), total_tokens=getattr(raw_usage, "input_tokens", 0) + getattr(raw_usage, "output_tokens", 0), ) return NormalizedResponse( content="".join(collected_text) or None, tool_calls=tool_calls or None, finish_reason=finish_reason, usage=usage, ) __all__ = [ "LLMClient", "api_mode", "complete", "stream", ]