Class UpstreamSemanticCompressor

uncommon_route/proxy.py:735–936

Runs semantic compression tasks through cheap upstream models.

class UpstreamSemanticCompressor:
    """Runs semantic compression tasks through cheap upstream models."""

    def __init__(
        self,
        *,
        upstream_chat: str,
        primary_api_key: str,
        providers_config: ProvidersConfig,
        model_mapper: ModelMapper,
        composition_policy: CompositionPolicy,
    ) -> None:
        self._upstream_chat = upstream_chat
        self._primary_api_key = primary_api_key
        self._providers = providers_config
        self._mapper = model_mapper
        self._policy = composition_policy

    def rebind_primary(
        self,
        *,
        upstream_chat: str,
        primary_api_key: str,
        model_mapper: ModelMapper,
    ) -> None:
        self._upstream_chat = upstream_chat
        self._primary_api_key = primary_api_key
        self._mapper = model_mapper

    def rebind_providers(self, providers_config: ProvidersConfig) -> None:
        self._providers = providers_config

    async def summarize_tool_result(
        self,
        content: str,
        *,
        tool_name: str,
        latest_user_prompt: str,
        request: Request,
    ) -> SemanticCallResult | None:
        system = (
            "You compress tool outputs for another model. Preserve facts, paths, errors, "
            "identifiers, counts, and anything actionable. Output plain text only."
        )
        user = (
            f"Latest user goal:\n{latest_user_prompt}\n\n"
            f"Tool: {tool_name or 'unknown'}\n"
            "Summarize the following tool result for continuation in under 220 words.\n\n"
            f"{content}"
        )
        return await self._run_task(
            request,
            self._policy.sidechannel.tool_summary,
            system,
            user,
            source_text=content,
            query_text=f"{latest_user_prompt} {tool_name}".strip(),
        )