Claude Code 接入自建开源模型：企业私有化与降本实践

# dynamic_tagging_handler.py
from litellm.integrations.custom_logger import CustomLogger
import litellm
class DynamicRoutingHandler(CustomLogger):
    def log_pre_api_call(self, kwargs, response_obj, start_time, end_time):
        """在 API 调用前拦截请求，修改路由目标"""
        # 1. 提取所有 message 的文本内容
        messages = kwargs.get("messages", [])
        full_text = self._extract_all_text(messages)
        # 2. 检测是否为 Hook 条件评估任务
        if self._is_hook_evaluator_prompt(full_text):
            print(f"[DynamicRouting] Detected hook evaluator prompt, routing to sagemaker-kimi-2-5")
            kwargs["model"] = "sagemaker-kimi-2-5"
        return kwargs
    def _extract_all_text(self, messages):
        """提取消息中的所有文本（支持多模态消息）"""
        text_parts = []
        for msg in messages:
            content = msg.get("content", "")
            if isinstance(content, str):
                text_parts.append(content)
            elif isinstance(content, list):
                # 处理多模态消息（文本 + 图片）
                for block in content:
                    if block.get("type") == "text":
                        text_parts.append(block.get("text", ""))
        return " ".join(text_parts)
    def _is_hook_evaluator_prompt(self, text):
        """检测是否为 Hook 条件评估 Prompt"""
        # 特征字符串列表
        markers = [
            "You are evaluating a hook in Claude Code",
            "hook condition",
            "Return your evaluation as a JSON object",
            '"satisfied": true'
        ]
        # 至少匹配 3 个特征才认为是该任务（避免误判）
        match_count = sum(1 for marker in markers if marker in text)
        return match_count >= 3
# 注册 Handler
proxy_handler_instance = DynamicRoutingHandler()

def _detect_task_type(self, text, messages):
    """综合判断任务类型"""
    if self._is_hook_evaluator(text):
        return "sagemaker-kimi-2-5"
    elif self._is_session_title_generator(text):
        return "sagemaker-kimi-2-5"
    elif self._is_bash_description_writer(text):
        return "sagemaker-kimi-2-5"
    elif len(text) > 10000:  # 长文本任务
        return "bedrock-claude-sonnet46"
    else:
        return None  # 保持原始模型

# stream_anthropic_schema_fixer.py
from litellm.integrations.custom_logger import CustomLogger
class AnthropicSchemaFixerHook(CustomLogger):
    async def async_post_call_streaming_iterator_hook(
        self,
        user_api_key_dict,
        response: AsyncGenerator,
        request_data: dict
    ) -> AsyncGenerator:
        """拦截流式响应，逐 chunk 修复 schema"""
        last_usage: Optional[Dict[str, Any]] = None
        chunk_count = 0
        async for chunk in response:
            chunk_count += 1
            # Pass through non-bytes chunks
            if not isinstance(chunk, bytes):
                yield chunk
                continue
            try:
                # Decode SSE format: "event: xxx\ndata: {...}\n\n"
                decoded = chunk.decode("utf-8")
                # Check if this is an SSE event with data
                if not decoded.startswith("event:") and not decoded.startswith("data:"):
                    yield chunk
                    continue
                # Parse SSE format
                event_type, data_json = self._parse_sse(decoded)
                # Apply fixes based on event type
                modified = False
                if event_type == "message_start":
                    modified = self._fix_message_start(data_json)
                elif event_type == "message_delta":
                    delta_modified, usage = self._fix_message_delta(data_json)
                    modified = delta_modified
                    # Track usage for message_stop
                    if usage:
                        last_usage = usage
                elif event_type == "message_stop":
                    modified = self._fix_message_stop(data_json, last_usage)
                # Re-encode if modified
                if modified:
                    yield self._rebuild_sse(event_type, data_json)
                else:
                    yield chunk
            except Exception as e:
                yield chunk
# 实例化并注册
hook = AnthropicSchemaFixerHook()