OpenWorker 深度解析（四）：LLM 層 — Provider 抽象、能力降級與 Context 自動壓縮

 1class ProviderClient(ABC):
 2    """Single-shot, provider-agnostic completion interface.
 3
 4    Deliberately blocking (the turn engine wraps it in `asyncio.to_thread`) and
 5    deliberately without a `max_turns` loop — the runtime owns the agent loop.
 6    """
 7
 8    @abstractmethod
 9    def complete(self, *, model, messages, tools=None, **settings) -> AssistantTurn: ...
10
11    @abstractmethod
12    def capabilities(self, model: str) -> ModelCapabilities: ...
13
14    def stream(self, *, model, messages, tools=None, **settings):
15        """Yield StreamChunks. Default: no token streaming — one final chunk with the
16        full turn. Providers that support streaming (OpenAIProvider) override this."""
17        yield StreamChunk(
18            turn=self.complete(model=model, messages=messages, tools=tools, **settings)
19        )

 1@dataclass
 2class AssistantTurn:
 3    text: Optional[str] = None
 4    tool_calls: list[ToolCall] = field(default_factory=list)
 5    finish_reason: Optional[str] = None
 6    raw: Any = field(default=None, repr=False, compare=False)
 7    # 模型的思考文字（DeepSeek reasoning_content、Gemini thought summaries…）
 8    # 顯示專用：存在 assistant 訊息的 `reasoning` 側車，但送 provider 前一律剝掉
 9    reasoning: Optional[str] = None
10    # Provider 私有側車（底線開頭 key，例如 `_gemini` thought signature）
11    # 契約：擁有它的 provider 會重新附加，其他 provider 必須剝掉或忽略
12    extras: dict[str, Any] = field(default_factory=dict)
13    # 正規化的 token 計數。provider 沒回報時是 None — 絕不猜測
14    usage: Optional[TokenUsage] = None

 1@dataclass
 2class TokenUsage:
 3    """Normalized token counts for one model round-trip.
 4
 5    `input` counts only fresh (uncached) prompt tokens; cached prompt tokens are
 6    split into `cache_read`/`cache_write`. Providers that don't report a cache
 7    split (Ollama, most compat vendors) leave the cache fields at 0. `output`
 8    includes thinking tokens where the vendor bills them as output (Gemini).
 9    """
10    input: int = 0
11    output: int = 0
12    cache_read: int = 0
13    cache_write: int = 0
14
15    @property
16    def context_tokens(self) -> int:
17        """Prompt-side total — what actually occupied the context window."""
18        return self.input + self.cache_read + self.cache_write

1return TokenUsage(
2    input=int(getattr(usage, "input_tokens", 0) or 0),          # 不含快取
3    output=int(getattr(usage, "output_tokens", 0) or 0),
4    cache_read=int(getattr(usage, "cache_read_input_tokens", 0) or 0),
5    cache_write=int(getattr(usage, "cache_creation_input_tokens", 0) or 0),
6)

1"""ProviderRouter — one `ProviderClient` that dispatches by the `provider:` prefix of a model
2string to a per-provider client, built lazily from its SecretStore profile and cached.
3
4This is the single provider the `SessionManager` hands to every engine, so `complete()/stream()`
5(which already receive the full model string per-call) route themselves: `ollama:llama3.3` →
6the Ollama client (Ollama's OpenAI-compatible `/v1`), bare `gpt-5.5` → the default (OpenAI).
7"""

 1@staticmethod
 2def _bare(model: str) -> str:
 3    """Strip a KNOWN provider prefix; the underlying SDK wants the bare model name. A model
 4    whose first segment isn't a provider (e.g. `qwen2.5-coder:32b` — a version tag, not a
 5    prefix) is returned unchanged, so the colon isn't mistaken for a provider separator."""
 6    if ":" in model:
 7        prefix, rest = model.split(":", 1)
 8        if get_descriptor(prefix) is not None:      # ← 必須是「已註冊的 provider」
 9            return rest
10    return model

 1def _client_for(self, model: str) -> ProviderClient:
 2    name = self._provider_name(model)
 3    with self._lock:
 4        client = self._clients.get(name)
 5        if client is None:
 6            profile = {}
 7            if self._secrets is not None:
 8                profile = self._secrets.get(f"provider:{name}") or {}
 9            client = build_provider_client(name, profile, self._secrets)
10            self._clients[name] = client
11        return client
12
13def invalidate(self, name: Optional[str] = None) -> None:
14    """Drop cached client(s) so the next call rebuilds with fresh config."""

1"""OpenAI Responses provider — native OpenAI models via `/v1/responses`.
2
3Chat Completions rejects function tools combined with any `reasoning_effort` other than
4`none` on GPT-5.6+ ("use /v1/responses"), which had reasoning pinned OFF for native OpenAI
5models. This provider is the Responses path: reasoning + tools at real effort levels,
6streamed reasoning summaries, and chain-of-thought continuity across tool round-trips via
7`store: false` + `include: ["reasoning.encrypted_content"]` — nothing retained server-side.
8"""

1def capabilities_for(model: str) -> ModelCapabilities:
2    # Curated models answer from the matrix (exact full-id match — including reseller ids
3    # like `together:zai-org/GLM-5.2`, whose names defeat the prefix heuristics below).
4    # Custom user-added models fall through to the heuristics, at their own risk.
5    entry = entry_for(model)
6    if entry is not None:
7        return entry.caps
8    ...  # 啟發式判斷

1"""The curated model matrix — the only models we actively suggest, label, and vouch for.
2
3Deliberately SMALL (owner call, 2026-07-04): current-generation, agent-capable (tool-calling)
4models only. It is not user-editable — users can still add any custom model string, which
5falls back to the conservative heuristics in ``capabilities.py`` at their own risk of
6degraded results.
7"""

1# Entries where the vendor spec wasn't re-checked stay ``None`` — the meter simply hides
2# rather than showing a made-up denominator.

 1# PDF attachments (stored as `file` parts) are adapted to the ACTIVE model right
 2# here — never in the persisted history — so a mid-session model switch always
 3# re-decides.
 4if any(p.get("type") == "file" for msg in out ... ):
 5    caps = self.provider.capabilities(self.model)
 6    if not getattr(caps, "pdf", False):
 7        from . import pdf_support
 8        out = [{**msg, "content": pdf_support.adapt_content(msg["content"], caps)} ...]
 9
10# 圖片同理
11if any(p.get("type") == "image_url" for msg in out ...):
12    caps = self.provider.capabilities(self.model)
13    if not getattr(caps, "vision", False):
14        placeholder = {"type": "text",
15                       "text": "[image attachment — not viewable by this model]"}

1if caps is not None and not getattr(caps, "vision", False) and self._history_has_images():
2    text += " — earlier images can't be read by this model"

 1"""Friendly translation of model access + quota failures.
 2
 3Matching is on the error BODY text (error codes/types), not just HTTP status — a 404 also
 4means "wrong base_url" and a 429 also means "slow down", and neither of those should be
 5dressed up as an access problem.
 6"""
 7
 8_NO_ACCESS = ("model_not_found", "does not exist or you do not have access",
 9              "does not have access to model", "permission_error", "permission denied")
10_NO_QUOTA = ("insufficient_quota", "exceeded your current quota",
11             "credit balance is too low", "billing hard limit")

1# Anthropic's 404 body is just "model: <id>" under type not_found_error; require both
2# halves so unrelated 404s (bad base_url, deleted resource) keep their raw message.
3if "not_found_error" in text and f"model: {model.split(':')[-1].lower()}" in text:
4    return no_access

 1def _add_cache_breakpoints(kwargs: dict[str, Any]) -> None:
 2    """Opt the request into prompt caching (5-minute ephemeral, prefix-matched).
 3
 4    Two breakpoints, the standard agent-loop shape:
 5    - last system block — caches tools + system together (tools render first);
 6    - last content block of the final message — caches the whole conversation
 7      prefix, so each request re-reads the previous turns' cache and writes only
 8      the new tail (append-only history keeps the prefix byte-identical).
 9
10    Outbound-only: the canonical history never carries `cache_control`.
11    Prefixes under the model's cacheable minimum silently don't cache; reads bill
12    ~0.1x and show up as `cache_read_input_tokens` (the metering's cache_read).
13    """
14    marker = {"type": "ephemeral"}
15    system = kwargs.get("system")
16    if isinstance(system, str) and system:
17        kwargs["system"] = [{"type": "text", "text": system, "cache_control": marker}]
18    messages = kwargs.get("messages") or []
19    if messages:
20        content = messages[-1].get("content")
21        if isinstance(content, list) and content:
22            content[-1] = {**content[-1], "cache_control": marker}

 1"""Auto-compaction of long session histories (OPE-27).
 2
 3When the outbound history approaches the model's context limit, the older portion of the
 4*outbound* view is replaced with (a) an LLM-written structured summary and (b) mechanically
 5extracted state — the recent turns and all user messages survive. The persisted transcript
 6is never modified; only what is sent to the model.
 7
 8This module is pure functions + one dataclass; the engine owns *when* (its run loop) and
 9*with what* (its provider/model), both injected here. That split keeps the engine.py
10footprint to a few lines and makes every policy testable without a provider.
11"""

1DEFAULT_THRESHOLD_PCT = 0.8
2DEFAULT_CAP_TOKENS = 250_000
3DEFAULT_CONTEXT_WINDOW = 128_000   # 矩陣裡沒有 context_window 時的預設
4KEEP_RECENT_FRACTION = 0.25
5SUMMARY_MAX_TOKENS = 3_000
6
7def trigger_tokens(context_window, *, threshold_pct=0.8, cap_tokens=250_000) -> int:
8    window = context_window or DEFAULT_CONTEXT_WINDOW
9    return min(int(threshold_pct * window), int(cap_tokens))

1# Trigger: min(threshold_pct × context_window, cap_tokens). The cap exists so 1M-context
2# models compact early — quality and latency degrade well before the nominal limit.

1def _compaction_due(self) -> bool:
2    cfg = self._compaction_config()
3    if cfg.get("enabled") is False:
4        return False
5    signal = self._last_context_tokens or _compaction.estimate_tokens(
6        self._outbound_messages()
7    )
8    return _compaction.should_compact(signal, cfg.get("context_window"), ...)

1def estimate_tokens(messages: list[dict[str, Any]]) -> int:
2    """chars/4 over the serialized messages — the fallback signal for providers that
3    never report usage (documented in the metering code)."""

1def pick_boundary(messages, *, keep_tokens: int) -> Optional[int]:
2    """The canonical index where the verbatim tail begins: the earliest turn start whose
3    suffix fits the keep budget. Prefers user-message boundaries; falls back to iteration
4    (assistant) boundaries when the newest turn alone exceeds the budget (a giant tool
5    loop). None when there is nothing meaningful to summarize."""

 1def _fit(candidates: list[int]) -> Optional[int]:
 2    for i in candidates:  # earliest-first: keep as much verbatim as fits
 3        if estimate_tokens(messages[i:]) <= keep_tokens:
 4            return i
 5    return None
 6
 7boundary = _fit(users)
 8if boundary is None and users:
 9    # The newest user turn alone blows the budget — cut inside it at an iteration
10    # boundary, keeping at least the most recent assistant step.
11    inside = [i for i in assistants if i > users[-1]]
12    boundary = _fit(inside)
13    if boundary is None:
14        boundary = inside[-1] if inside else users[-1]

 1SUMMARY_SYSTEM_PROMPT = """You are compacting an AI coworker's session history so the coworker
 2can continue working in a smaller context. Write a structured summary of the conversation below.
 3It is the coworker's ONLY memory of these turns, so preserve everything load-bearing.
 4...
 51. **Primary request and intent** — … including standing constraints stated at any point
 6   (e.g. "never send without my approval"). Constraints outlive the turns they were stated in.
 72. **Key concepts and decisions** — … Include the WHY, not just the what — a decision without
 8   its reason gets relitigated.
 9...
10Rules:
11- Do NOT carry full file contents as truth. Note THAT a file was read/edited; the coworker
12  re-reads if it needs the content again. Stale memory of a file is worse than no memory.
13- Be concrete: paths, names, commands, ids — not vague references.
14- Output only the summary sections, no preamble."""

1CONTINUATION_CONTRACT = (
2    "Continue where you left off: pick up the current work and next step exactly as "
3    "described. Do not re-ask answered questions, do not recap, do not mention that the "
4    "context was compacted. If you need the contents of a file noted above, re-read it."
5)

1_SPAN_TOOL_RESULT_CLIP = 400      # 每則工具結果最多 400 字
2_SPAN_BUDGET_CHARS = 400_000      # 整段 render 的上限
3
4def _render_span(span, *, budget_chars=_SPAN_BUDGET_CHARS) -> str:
5    """The summarized span as compact text for the summarizer. Tool results are clipped
6    hard (first casualty); if the whole render still exceeds the budget, oldest lines are
7    dropped — the newest context is the most load-bearing."""

1def summarizer_messages(span, *, prior_summary: str = "") -> list[dict]:
2    """On repeated compaction the previous summary is message zero of the new span —
3    summarized along with the turns since."""
4    body = _render_span(span)
5    if prior_summary:
6        body = ("[previous compaction summary — fold its still-relevant content into the new "
7                "summary]\n" + prior_summary + "\n\n[conversation since]\n" + body)

1_USER_MESSAGE_CLIP = 600
2_USER_MESSAGES_MAX = 40
3# User messages preserved mechanically in the compacted block ("trimmed of pasted bulk").
4# The list is capped to the newest N across repeated compactions — otherwise it appends
5# forever and the block slowly reclaims the window it freed. Dropped ones stay counted
6# (their intent lives in the summary, which is asked to list user messages too).

1if state.user_messages_dropped:
2    parts += [f"({state.user_messages_dropped} earlier user messages omitted — "
3              "their intent is covered by the summary above)"]

 1def apply_to_outbound(messages, state) -> list[dict]:
 2    """The outbound view: [system?] + the compacted block (as a user message) + the
 3    verbatim tail. Canonical history is untouched; provider-private sidecars in the
 4    summarized span vanish with it (replay chains legally restart after a compaction
 5    point). No-op when state is absent or stale."""
 6    if state is None:
 7        return messages
 8    boundary = state.boundary_index
 9    if boundary <= 0 or boundary >= len(messages):
10        return messages
11    head = []
12    if messages and messages[0].get("role") == "system":
13        head.append(messages[0])
14    head.append({"role": "user", "content": compacted_block(state)})
15    return head + messages[boundary:]

1# Auto-compaction (OPE-27): everything before the boundary is represented by the
2# compacted block. Outbound-only — the canonical history stays intact — and the
3# block+tail are byte-stable between turns, so prompt caching keeps working.