From 14ce901bc2c59b9d75ce1200395f848ae149f1e4 Mon Sep 17 00:00:00 2001 From: ThreeFish Date: Sat, 18 Apr 2026 12:28:51 +0800 Subject: [PATCH] =?UTF-8?q?refactor(vendor-channels):=20=E5=B0=86=20vendor?= =?UTF-8?q?=20=E8=BD=AC=E6=8D=A2=E4=B8=93=E5=B1=9E=E5=87=BD=E6=95=B0?= =?UTF-8?q?=E4=BB=8E=20request=5Fnormalizer=20=E8=BF=81=E5=85=A5=20vendor?= =?UTF-8?q?=5Fchannels=20=E7=BB=9F=E4=B8=80=E7=BB=B4=E6=8A=A4;?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除 request_normalizer.py 中已废弃的 Phase 2 逻辑（apply_anthropic_specific_fixes、_repair_orphaned_tool_use 等）， 将 enforce_anthropic_tool_pairing 和 strip_thinking_blocks 移入 vendor_channels.py， 合并重复的 _strip_thinking_blocks_inplace 为统一的 strip_thinking_blocks 公开函数， 移除 normalization 参数在 routes→router→executor 三层透传中的死参数链， 精简 NormalizationResult 仅保留 Phase 1 必要字段。 🤖 Generated with [Claude Code](https://github.com/claude), [CodeX](https://openai.com), [Gemini](https://github.com/apps/gemini-code-assist) Co-Authored-By: Aurelius Huang --- src/coding/proxy/convert/vendor_channels.py | 155 +++- src/coding/proxy/routing/executor.py | 2 - src/coding/proxy/routing/router.py | 10 +- src/coding/proxy/server/request_normalizer.py | 531 +------------ src/coding/proxy/server/routes.py | 16 +- tests/test_request_normalizer.py | 743 +----------------- tests/test_vendor_channels.py | 20 +- 7 files changed, 192 insertions(+), 1285 deletions(-) diff --git a/src/coding/proxy/convert/vendor_channels.py b/src/coding/proxy/convert/vendor_channels.py index e41f842..0c9146c 100644 --- a/src/coding/proxy/convert/vendor_channels.py +++ b/src/coding/proxy/convert/vendor_channels.py @@ -40,11 +40,13 @@ def get_transition_channel( # ── 共享辅助函数 ────────────────────────────────────────────── -def _strip_thinking_blocks_inplace(body: dict[str, Any]) -> int: +def strip_thinking_blocks(body: dict[str, Any]) -> int: """从 assistant 消息中移除 thinking/redacted_thinking 块（就地）. - GLM-5 等 Anthropic 兼容端点不支持非 Anthropic 签发的 thinking signature， - 跨供应商场景中必须剥离以避免 400 invalid_request_error。 + Anthropic API 要求 thinking blocks 的 signature 必须是其签发的有效签名。 + 跨供应商迁移（如 Zhipu → Anthropic）后，conversation history 中可能包含 + 非 Anthropic 签发的 signature，导致 400 invalid_request_error。 + 根据 Anthropic 官方文档，thinking blocks 可以被安全省略，不影响模型行为。 剥离后 content 为空时插入最小占位 text block 以保持消息结构合法性。 @@ -69,11 +71,145 @@ def _strip_thinking_blocks_inplace(body: dict[str, Any]) -> int: removed = original_len - len(new_content) if removed and not new_content: new_content = [{"type": "text", "text": "[thinking]"}] + logger.info( + "Inserted placeholder text block after stripping " + "%d thinking block(s) to avoid empty assistant content", + removed, + ) message["content"] = new_content stripped += removed return stripped +def enforce_anthropic_tool_pairing( + messages_list: list[dict[str, Any]], +) -> list[str]: + """为跨供应商场景强制保证 Anthropic tool_use/tool_result 配对约束. + + 单次正向遍历所有消息，对每个 assistant 消息执行： + + 1. 剥离所有 tool_result 块（跨供应商产物，如 GLM-5 内联的 tool_result） + 2. 收集所有 tool_use ID + 3. 确保紧邻的下一条消息是 user 消息且包含所有必需的 tool_result + 4. 将剥离的 tool_result 重定位到正确的 user 消息 + 5. 为仍缺失的 tool_result 合成 ``is_error=True`` 的占位块 + + 此函数是一个**自包含的单遍处理**，不依赖 Phase 1 收集的 misplaced 信息。 + + Args: + messages_list: 消息列表（就地修改）。 + + Returns: + 新增的 adaptation 标签列表。 + """ + adaptations: list[str] = [] + relocated_count = 0 + synthesized_ids: list[str] = [] + + i = 0 + while i < len(messages_list): + msg = messages_list[i] + if not isinstance(msg, dict) or msg.get("role") != "assistant": + i += 1 + continue + + content = msg.get("content") + if not isinstance(content, list): + i += 1 + continue + + # A. 从 assistant 消息中剥离所有 tool_result 块 + extracted_tool_results: dict[str, dict[str, Any]] = {} + retained_content: list[Any] = [] + for block in content: + if isinstance(block, dict) and block.get("type") == "tool_result": + tid = block.get("tool_use_id") + if tid: + extracted_tool_results[tid] = block + relocated_count += 1 + else: + retained_content.append(block) + + if extracted_tool_results: + msg["content"] = retained_content + + # B. 收集所有 tool_use ID + tool_use_ids: list[str] = [ + b["id"] + for b in ( + msg.get("content") if isinstance(msg.get("content"), list) else [] + ) + if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("id") + ] + if not tool_use_ids: + current_content = msg.get("content") + if isinstance(current_content, list) and not current_content: + msg["content"] = [{"type": "text", "text": ""}] + i += 1 + continue + + # C. 确保 messages[i+1] 是 user 消息 + next_idx = i + 1 + if ( + next_idx < len(messages_list) + and isinstance(messages_list[next_idx], dict) + and messages_list[next_idx].get("role") == "user" + ): + user_msg = messages_list[next_idx] + else: + user_msg: dict[str, Any] = {"role": "user", "content": []} + messages_list.insert(next_idx, user_msg) + + # D. 确保 user_msg.content 是 list + user_content = user_msg.get("content") + if isinstance(user_content, str): + user_msg["content"] = [{"type": "text", "text": user_content}] + elif not isinstance(user_content, list): + user_msg["content"] = [] + + # E. 收集 user 消息中已有的 tool_result IDs + existing_result_ids: set[str] = { + b["tool_use_id"] + for b in user_msg["content"] + if isinstance(b, dict) + and b.get("type") == "tool_result" + and b.get("tool_use_id") + } + + # F. 为每个 tool_use_id 确保 tool_result 存在 + for uid in tool_use_ids: + if uid in existing_result_ids: + continue + if uid in extracted_tool_results: + user_msg["content"].append(extracted_tool_results[uid]) + else: + user_msg["content"].append( + { + "type": "tool_result", + "tool_use_id": uid, + "content": "", + "is_error": True, + } + ) + synthesized_ids.append(uid) + + i += 1 + + if relocated_count: + adaptations.append("misplaced_tool_result_relocated") + if synthesized_ids: + adaptations.append("orphaned_tool_use_repaired") + logger.warning( + "Vendor degradation adaptation: synthesized %d tool_result block(s) " + "for orphaned tool_use to satisfy Anthropic pairing constraint. " + "Affected tool_use_ids: %s", + len(synthesized_ids), + ", ".join(synthesized_ids), + ) + + return adaptations + + def _strip_cache_control(body: dict[str, Any]) -> int: """从 system/messages/tools 中移除 cache_control 字段（就地）. @@ -135,7 +271,7 @@ def prepare_copilot_to_zhipu( adaptations: list[str] = [] # Step 1: 剥离 thinking/redacted_thinking 块 - stripped = _strip_thinking_blocks_inplace(prepared) + stripped = strip_thinking_blocks(prepared) if stripped: adaptations.append(f"stripped_{stripped}_thinking_blocks") @@ -151,8 +287,6 @@ def prepare_copilot_to_zhipu( adaptations.append(f"removed_{param}_param") # Step 4: 强制 tool_use/tool_result 配对 - from ..server.request_normalizer import enforce_anthropic_tool_pairing - pairing_fixes = enforce_anthropic_tool_pairing(prepared.get("messages", [])) if pairing_fixes: adaptations.extend(pairing_fixes) @@ -183,7 +317,7 @@ def prepare_zhipu_to_copilot( adaptations: list[str] = [] # Step 1: 剥离 thinking/redacted_thinking 块 - stripped = _strip_thinking_blocks_inplace(prepared) + stripped = strip_thinking_blocks(prepared) if stripped: adaptations.append(f"stripped_{stripped}_thinking_blocks") @@ -193,8 +327,6 @@ def prepare_zhipu_to_copilot( adaptations.append(f"removed_{removed_cc}_cache_control_fields") # Step 3: 强制 tool_use/tool_result 配对 - from ..server.request_normalizer import enforce_anthropic_tool_pairing - pairing_fixes = enforce_anthropic_tool_pairing(prepared.get("messages", [])) if pairing_fixes: adaptations.extend(pairing_fixes) @@ -223,11 +355,6 @@ def prepare_zhipu_to_anthropic( Returns: (prepared_body, adaptations) — adaptations 为应用的变换描述列表。 """ - from ..server.request_normalizer import ( - enforce_anthropic_tool_pairing, - strip_thinking_blocks, - ) - prepared = copy.deepcopy(body) adaptations: list[str] = [] diff --git a/src/coding/proxy/routing/executor.py b/src/coding/proxy/routing/executor.py index 7bd52ac..d9f84b4 100644 --- a/src/coding/proxy/routing/executor.py +++ b/src/coding/proxy/routing/executor.py @@ -283,7 +283,6 @@ async def execute_stream( self, body: dict[str, Any], headers: dict[str, str], - normalization: Any = None, ) -> AsyncIterator[tuple[bytes, str]]: """路由流式请求，按优先级尝试各层级.""" last_idx = len(self._tiers) - 1 @@ -453,7 +452,6 @@ async def execute_message( self, body: dict[str, Any], headers: dict[str, str], - normalization: Any = None, ) -> VendorResponse: """路由非流式请求，按优先级尝试各层级.""" last_idx = len(self._tiers) - 1 diff --git a/src/coding/proxy/routing/router.py b/src/coding/proxy/routing/router.py index 355a91f..3a65cd6 100644 --- a/src/coding/proxy/routing/router.py +++ b/src/coding/proxy/routing/router.py @@ -134,24 +134,18 @@ async def route_stream( self, body: dict[str, Any], headers: dict[str, str], - normalization: Any = None, ) -> AsyncIterator[tuple[bytes, str]]: """路由流式请求，按优先级尝试各层级.""" - async for chunk, vendor_name in self._executor.execute_stream( - body, headers, normalization=normalization - ): + async for chunk, vendor_name in self._executor.execute_stream(body, headers): yield chunk, vendor_name async def route_message( self, body: dict[str, Any], headers: dict[str, str], - normalization: Any = None, ) -> Any: """路由非流式请求，按优先级尝试各层级.""" - return await self._executor.execute_message( - body, headers, normalization=normalization - ) + return await self._executor.execute_message(body, headers) # ── 生命周期 ─────────────────────────────────────────── diff --git a/src/coding/proxy/server/request_normalizer.py b/src/coding/proxy/server/request_normalizer.py index e445870..cbc18f5 100644 --- a/src/coding/proxy/server/request_normalizer.py +++ b/src/coding/proxy/server/request_normalizer.py @@ -10,27 +10,12 @@ logger = logging.getLogger(__name__) -# ── 跨请求日志去重：记录已报告过的 misplaced tool_use_id ────────── -# 同一 tool_use_id 仅首次输出 WARNING（含完整因果上下文）， -# 后续在同一会话中重复出现时降级为 DEBUG，避免日志噪声。 -_LOGGED_MISPLACED_TOOL_IDS: set[str] = set() -_LOGGED_MISPLACED_TOOL_IDS_MAX = 500 - _ANTHROPIC_TOOL_USE_ID_RE = re.compile(r"^toolu_[A-Za-z0-9_]+$") _ANTHROPIC_SERVER_TOOL_USE_ID_RE = re.compile(r"^srvtoolu_[A-Za-z0-9_]+$") _VENDOR_TOOL_BLOCK_TYPES = { "server_tool_use_delta", } -# 标识跨供应商迁移产物的 adaptation 名称集合（用于条件化 thinking block 剥离） -_CROSS_VENDOR_ADAPTATIONS = frozenset( - { - "vendor_block_removed", - "server_tool_use_id_rewritten_for_anthropic", - "invalid_tool_use_id_rewritten_for_anthropic", - } -) - @dataclass class NormalizationResult: @@ -39,52 +24,20 @@ class NormalizationResult: body: dict[str, Any] adaptations: list[str] = field(default_factory=list) fatal_reasons: list[str] = field(default_factory=list) - # Phase 2 上下文（仅 Anthropic tier 使用） - tool_id_map: dict[str, str] = field(default_factory=dict) - misplaced_tool_results: list[tuple[int, dict[str, Any]]] = field( - default_factory=list - ) - misplaced_log_info: list[tuple[str, int, int, str]] = field(default_factory=list) @property def recoverable(self) -> bool: return not self.fatal_reasons - @property - def has_anthropic_fixes(self) -> bool: - """是否需要应用 Anthropic 专属修复（重定位 + 孤儿修复）.""" - return bool(self.misplaced_tool_results) or bool(self.tool_id_map) - - @property - def has_cross_vendor_signals(self) -> bool: - """是否检测到跨供应商迁移信号（用于条件化 thinking block 剥离）. - - 当请求体中包含非 Anthropic 原生格式的产物（如非标准 tool_use ID、 - 供应商私有块、错位的 tool_result 等）时返回 True。 - """ - if self.tool_id_map: - return True - if self.misplaced_tool_results: - return True - return any( - any(a.startswith(prefix) for prefix in _CROSS_VENDOR_ADAPTATIONS) - for a in self.adaptations - ) - def normalize_anthropic_request(body: dict[str, Any]) -> NormalizationResult: """清洗供应商私有块，尽量恢复为合法 Anthropic Messages 请求. - 这是 vendor-agnostic 的 Phase 1 规范化：对所有 vendor 均适用。 + vendor-agnostic 的 Phase 1 规范化：对所有 vendor 均适用。 处理策略： 1. 移除供应商私有块（如 server_tool_use_delta） 2. 重写无效/非标准的 tool_use / tool_result ID - 3. **收集**（但不应用）错位的 tool_result 块信息，供 Phase 2 使用 - - Phase 2（Anthropic 专属修复：重定位 + 孤儿修复）由 - :func:`apply_anthropic_specific_fixes` 独立执行，仅在请求实际发送给 - Anthropic tier 时调用，确保 Zhipu 等其他 vendor 不受影响。 """ normalized = copy.deepcopy(body) adaptations: list[str] = [] @@ -97,14 +50,6 @@ def next_tool_id() -> str: normalized_counter += 1 return f"toolu_normalized_{normalized_counter}" - # 收集本轮 misplaced tool_result 块（Phase 2 延迟到 Anthropic tier 执行） - collected_misplaced: list[ - tuple[int, dict[str, Any]] - ] = [] # (source_msg_idx, block) - misplaced_log_info: list[ - tuple[str, int, int, str] - ] = [] # (role, msg_idx, blk_idx, tool_use_id) - def normalize_content_block( block: Any, *, @@ -166,8 +111,6 @@ def normalize_content_block( _ANTHROPIC_TOOL_USE_ID_RE.match(tool_use_id) or _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(tool_use_id) ): - # 保持原样。对 server_tool_use_id 的用户结果，若未在当前请求体中出现， - # 交由上游决定是否接受，避免错误猜测跨轮次关联。 return normalized_block elif isinstance(tool_use_id, str) and tool_use_id: fatal_reasons.append( @@ -181,23 +124,14 @@ def normalize_content_block( return None return normalized_block - # tool_result 出现在非 user 消息中（如 assistant）—— 仅收集供 Phase 2 使用。 - # Phase 2 由 apply_anthropic_specific_fixes() 执行，仅在 Anthropic tier 时调用。 - # 对于 Zhipu 等其他 vendor，misplaced 块保留在原位不变。 + # tool_result 出现在非 user 消息中（跨供应商产物）—— 保留原样。 + # 跨供应商的 tool_use/tool_result 配对修复由 vendor_channels.py 中的 + # enforce_anthropic_tool_pairing() 在源→目标转换通道中处理。 normalized_block = dict(block) tool_use_id = normalized_block.get("tool_use_id") if isinstance(tool_use_id, str) and tool_use_id in tool_id_map: normalized_block["tool_use_id"] = tool_id_map[tool_use_id] adaptations.append("tool_result_tool_use_id_rewritten") - collected_misplaced.append((message_index, normalized_block)) - misplaced_log_info.append( - ( - message_role, - message_index, - block_index, - normalized_block.get("tool_use_id", "N/A"), - ) - ) return normalized_block return dict(block) @@ -225,461 +159,4 @@ def normalize_content_block( body=normalized, adaptations=sorted(set(adaptations)), fatal_reasons=fatal_reasons, - tool_id_map=tool_id_map, - misplaced_tool_results=collected_misplaced, - misplaced_log_info=misplaced_log_info, ) - - -def apply_anthropic_specific_fixes( - messages_list: list[dict[str, Any]], - misplaced_results: list[tuple[int, dict[str, Any]]], - misplaced_log_info: list[tuple[str, int, int, str]], -) -> list[str]: - """应用 Anthropic 专属修复（重定位 + 孤儿修复）. - - 仅在请求实际发送给 Anthropic tier 时调用，确保 Zhipu 等其他 vendor 不受影响。 - Phase 1（normalize_anthropic_request）仅收集 misplaced 信息，将实际修复延迟到此函数。 - - Args: - messages_list: 消息列表（就地修改）。 - misplaced_results: Phase 1 收集的 misplaced tool_result 列表， - 每个元素为 (source_msg_idx, block)。 - misplaced_log_info: Phase 1 收集的日志信息列表， - 每个元素为 (role, msg_idx, blk_idx, tool_use_id)。 - - Returns: - 新增的 adaptation 标签列表。 - """ - adaptations: list[str] = [] - - if misplaced_results: - # ── 1. 从源消息中移除 misplaced tool_result 块 ─────────── - to_remove: dict[int, set[str]] = {} - for source_idx, block in misplaced_results: - tid = block.get("tool_use_id", "") - if tid: - to_remove.setdefault(source_idx, set()).add(tid) - - for msg_idx, tids in to_remove.items(): - if msg_idx < len(messages_list): - msg = messages_list[msg_idx] - if isinstance(msg, dict): - content = msg.get("content") - if isinstance(content, list): - msg["content"] = [ - b - for b in content - if not ( - isinstance(b, dict) - and b.get("type") == "tool_result" - and b.get("tool_use_id") in tids - ) - ] - - # ── 2. 重定位到紧邻的 user 消息 ────────────────────────── - # 按源消息索引降序处理，避免插入新消息时索引偏移。 - for source_idx, result_block in sorted( - misplaced_results, key=lambda x: x[0], reverse=True - ): - target_user_idx = None - for j in range(source_idx + 1, len(messages_list)): - if ( - isinstance(messages_list[j], dict) - and messages_list[j].get("role") == "user" - ): - target_user_idx = j - break - - if target_user_idx is not None: - target_content = messages_list[target_user_idx].get("content") - if isinstance(target_content, list): - target_content.append(result_block) - elif isinstance(target_content, str): - # string content 转为 text block 后追加，避免丢失原始文本 - messages_list[target_user_idx]["content"] = [ - {"type": "text", "text": target_content}, - result_block, - ] - else: - messages_list[target_user_idx]["content"] = [result_block] - else: - # 无后续 user 消息：插入一条合成 user 消息 - messages_list.insert( - source_idx + 1, - { - "role": "user", - "content": [result_block], - }, - ) - - adaptations.append("misplaced_tool_result_relocated") - - if misplaced_log_info: - _emit_misplaced_tool_result_summary(misplaced_log_info) - - # ── 3. 修复通道：为孤儿 tool_use 合成 tool_result ────────────── - repaired = _repair_orphaned_tool_use(messages_list) - if repaired: - adaptations.append("orphaned_tool_use_repaired") - total_synthesized = sum(repaired.values()) - logger.warning( - "Vendor degradation adaptation: synthesized %d tool_result block(s) " - "for orphaned tool_use to satisfy Anthropic pairing constraint. " - "Affected tool_use_ids: %s", - total_synthesized, - ", ".join(sorted(repaired)), - ) - - return adaptations - - -def _emit_misplaced_tool_result_summary( - stripped: list[tuple[str, int, int, str]], -) -> None: - """为被重定位的 misplaced tool_result 输出汇总日志. - - 策略： - - 将同一次请求中的多个重定位事件合并为单条日志 - - 首次出现的 tool_use_id → WARNING（含完整因果上下文） - - 同一 tool_use_id 在后续请求中再次出现 → DEBUG（避免日志噪声） - - 注意：此函数运行在 asyncio 事件循环的主线程中，set 操作无需加锁。 - - Args: - stripped: 每个元素为 (message_role, message_index, block_index, tool_use_id) - """ - # 提取去重后的 tool_use_id 集合 - unique_tool_ids = {tid for _, _, _, tid in stripped} - - # 区分首次出现 vs 已报告过的 - new_id_set = unique_tool_ids - _LOGGED_MISPLACED_TOOL_IDS - known_id_set = unique_tool_ids & _LOGGED_MISPLACED_TOOL_IDS - - # 更新已报告集合（防止无限增长：保留最近一半条目） - _LOGGED_MISPLACED_TOOL_IDS.update(unique_tool_ids) - if len(_LOGGED_MISPLACED_TOOL_IDS) > _LOGGED_MISPLACED_TOOL_IDS_MAX: - to_keep = sorted(_LOGGED_MISPLACED_TOOL_IDS)[ - _LOGGED_MISPLACED_TOOL_IDS_MAX // 2 : - ] - _LOGGED_MISPLACED_TOOL_IDS.clear() - _LOGGED_MISPLACED_TOOL_IDS.update(to_keep) - - if new_id_set: - # 首次出现：WARNING + 完整因果上下文 - positions = ", ".join( - f"messages.{mi}.content.{bi} (role={r}, tool_use_id={tid})" - for r, mi, bi, tid in stripped - if tid in new_id_set - ) - new_count = sum(1 for _, _, _, tid in stripped if tid in new_id_set) - logger.warning( - "Vendor degradation adaptation: relocated %d misplaced tool_result block(s) " - "from non-user message(s) to adjacent user message(s). Cause: cross-vendor " - "conversation history contains tool_result blocks in assistant messages " - "(typical when GLM-5 includes tool results inline in responses). Anthropic " - "API requires tool_result only in user messages, so these blocks are " - "relocated to maintain tool_use/tool_result pairing. Affected: %s. " - "Subsequent occurrences of these tool_use_ids will be logged at DEBUG level.", - new_count, - positions, - ) - - if known_id_set: - # 已报告过的：DEBUG - known_count = sum(1 for _, _, _, tid in stripped if tid in known_id_set) - logger.debug( - "Normalization: relocated %d previously reported misplaced tool_result " - "block(s) (tool_use_ids: %s)", - known_count, - ", ".join(sorted(known_id_set)), - ) - - -def _repair_orphaned_tool_use( - messages_list: list[dict[str, Any]], -) -> dict[str, int]: - """为孤儿 tool_use 块合成缺失的 tool_result. - - 遍历所有 assistant 消息，检查紧邻的 user 消息是否包含每个 tool_use - 对应的 tool_result。对于缺失的 tool_result，合成一个 ``is_error=true`` - 的占位块以满足 Anthropic API 约束。 - - Args: - messages_list: 消息列表（就地修改）。 - - Returns: - dict: 以 tool_use_id 为键、修复次数为值的映射；无修复时返回空 dict。 - """ - repaired: dict[str, int] = {} - i = 0 - while i < len(messages_list): - msg = messages_list[i] - if not isinstance(msg, dict) or msg.get("role") != "assistant": - i += 1 - continue - - content = msg.get("content") - if not isinstance(content, list): - i += 1 - continue - - # 收集当前 assistant 消息中所有 tool_use 的 id - tool_use_ids: list[str] = [ - b["id"] - for b in content - if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("id") - ] - if not tool_use_ids: - i += 1 - continue - - # 检查紧邻的 user 消息中已有的 tool_result - next_msg = messages_list[i + 1] if i + 1 < len(messages_list) else None - existing_result_ids: set[str] = set() - - if isinstance(next_msg, dict) and next_msg.get("role") == "user": - next_content = next_msg.get("content") - if isinstance(next_content, list): - existing_result_ids = { - b["tool_use_id"] - for b in next_content - if isinstance(b, dict) - and b.get("type") == "tool_result" - and b.get("tool_use_id") - } - - # 找出缺失的 tool_result - orphan_ids = [uid for uid in tool_use_ids if uid not in existing_result_ids] - if not orphan_ids: - i += 1 - continue - - # 为每个孤儿 tool_use 合成 tool_result - synthetic_blocks = [ - { - "type": "tool_result", - "tool_use_id": uid, - "content": "", - "is_error": True, - } - for uid in orphan_ids - ] - - if isinstance(next_msg, dict) and next_msg.get("role") == "user": - next_content = next_msg.get("content") - if isinstance(next_content, list): - next_content.extend(synthetic_blocks) - elif isinstance(next_content, str): - next_msg["content"] = [ - {"type": "text", "text": next_content} - ] + synthetic_blocks - else: - next_msg["content"] = synthetic_blocks - else: - # 无紧邻 user 消息：插入合成 user 消息 - messages_list.insert(i + 1, {"role": "user", "content": synthetic_blocks}) - - for uid in orphan_ids: - repaired[uid] = repaired.get(uid, 0) + 1 - - i += 1 - return repaired - - -# ── Phase 2: 跨供应商 tool_use/tool_result 配对强制修复 ───────── - - -def enforce_anthropic_tool_pairing( - messages_list: list[dict[str, Any]], -) -> list[str]: - """为跨供应商场景强制保证 Anthropic tool_use/tool_result 配对约束. - - 单次正向遍历所有消息，对每个 assistant 消息执行： - - 1. 剥离所有 tool_result 块（跨供应商产物，如 GLM-5 内联的 tool_result） - 2. 收集所有 tool_use ID - 3. 确保紧邻的下一条消息是 user 消息且包含所有必需的 tool_result - 4. 将剥离的 tool_result 重定位到正确的 user 消息 - 5. 为仍缺失的 tool_result 合成 ``is_error=True`` 的占位块 - - 此函数是一个**自包含的单遍处理**，不依赖 Phase 1 收集的 misplaced 信息， - 通过直接扫描消息列表确保处理的完备性。替代此前多步串联管线 - （剥离 → 重定位 → 孤儿修复）因步骤间隐式依赖导致的漏修问题。 - - 仅在请求实际发送给 Anthropic tier 且检测到跨供应商信号时调用， - 确保 Zhipu 等其他 vendor 不受影响。 - - Args: - messages_list: 消息列表（就地修改）。 - - Returns: - 新增的 adaptation 标签列表。 - """ - adaptations: list[str] = [] - relocated_count = 0 - synthesized_ids: list[str] = [] - - i = 0 - while i < len(messages_list): - msg = messages_list[i] - if not isinstance(msg, dict) or msg.get("role") != "assistant": - i += 1 - continue - - content = msg.get("content") - if not isinstance(content, list): - i += 1 - continue - - # ── A. 从 assistant 消息中剥离所有 tool_result 块 ───────── - extracted_tool_results: dict[str, dict[str, Any]] = {} # tool_use_id → block - retained_content: list[Any] = [] - for block in content: - if isinstance(block, dict) and block.get("type") == "tool_result": - tid = block.get("tool_use_id") - if tid: - extracted_tool_results[tid] = block - relocated_count += 1 - # 无 tool_use_id 的 tool_result 直接丢弃（无效块） - else: - retained_content.append(block) - - if extracted_tool_results: - msg["content"] = retained_content - - # ── B. 收集所有 tool_use ID ─────────────────────────────── - tool_use_ids: list[str] = [ - b["id"] - for b in ( - msg.get("content") if isinstance(msg.get("content"), list) else [] - ) - if isinstance(b, dict) and b.get("type") == "tool_use" and b.get("id") - ] - if not tool_use_ids: - # 无 tool_use 块：若剥离后 content 为空，插入占位 - current_content = msg.get("content") - if isinstance(current_content, list) and not current_content: - msg["content"] = [{"type": "text", "text": ""}] - i += 1 - continue - - # ── C. 确保 messages[i+1] 是 user 消息 ─────────────────── - next_idx = i + 1 - if ( - next_idx < len(messages_list) - and isinstance(messages_list[next_idx], dict) - and messages_list[next_idx].get("role") == "user" - ): - user_msg = messages_list[next_idx] - else: - # 插入合成 user 消息 - user_msg: dict[str, Any] = {"role": "user", "content": []} - messages_list.insert(next_idx, user_msg) - - # ── D. 确保 user_msg.content 是 list ───────────────────── - user_content = user_msg.get("content") - if isinstance(user_content, str): - user_msg["content"] = [{"type": "text", "text": user_content}] - elif not isinstance(user_content, list): - user_msg["content"] = [] - - # ── E. 收集 user 消息中已有的 tool_result IDs ───────────── - existing_result_ids: set[str] = { - b["tool_use_id"] - for b in user_msg["content"] - if isinstance(b, dict) - and b.get("type") == "tool_result" - and b.get("tool_use_id") - } - - # ── F. 为每个 tool_use_id 确保 tool_result 存在 ────────── - for uid in tool_use_ids: - if uid in existing_result_ids: - continue # 已有匹配的 tool_result - - if uid in extracted_tool_results: - # 从 assistant 剥离的 tool_result 重定位到 user - user_msg["content"].append(extracted_tool_results[uid]) - else: - # 完全缺失：合成 is_error=True 占位块 - user_msg["content"].append( - { - "type": "tool_result", - "tool_use_id": uid, - "content": "", - "is_error": True, - } - ) - synthesized_ids.append(uid) - - i += 1 - - # ── 构建 adaptation 标签与日志 ──────────────────────────── - if relocated_count: - adaptations.append("misplaced_tool_result_relocated") - if synthesized_ids: - adaptations.append("orphaned_tool_use_repaired") - logger.warning( - "Vendor degradation adaptation: synthesized %d tool_result block(s) " - "for orphaned tool_use to satisfy Anthropic pairing constraint. " - "Affected tool_use_ids: %s", - len(synthesized_ids), - ", ".join(synthesized_ids), - ) - - return adaptations - - -# ── Phase 2: Thinking block 剥离 ────────────────────────────── - -# 需要从 assistant messages 中剥离的 thinking block 类型 -_THINKING_BLOCK_TYPES = {"thinking", "redacted_thinking"} - - -def strip_thinking_blocks(body: dict[str, Any]) -> int: - """从 assistant messages 中移除 thinking / redacted_thinking blocks. - - Anthropic API 要求 thinking blocks 的 ``signature`` 必须是其签发的有效签名。 - 跨供应商迁移（如 Zhipu → Anthropic）后，conversation history 中可能包含 - 非 Anthropic 签发的 signature，导致 400 ``invalid_request_error``。 - 根据 Anthropic 官方文档，thinking blocks 可以被安全省略，不影响模型行为。 - - 此函数仅在 executor 检测到跨供应商信号时调用（条件化剥离）， - 纯 Anthropic 会话中 thinking blocks 保持原样以维持上下文连续性。 - - Args: - body: 请求体（就地修改）。 - - Returns: - 被移除的 thinking block 数量。 - """ - stripped = 0 - for message in body.get("messages", []): - if not isinstance(message, dict) or message.get("role") != "assistant": - continue - content = message.get("content") - if not isinstance(content, list): - continue - original_len = len(content) - new_content = [ - block - for block in content - if not ( - isinstance(block, dict) and block.get("type") in _THINKING_BLOCK_TYPES - ) - ] - removed = original_len - len(new_content) - if removed and not new_content: - # 剥离所有 thinking blocks 后 content 为空 —— - # Anthropic API 要求非末尾 assistant message 的 content 必须非空。 - # 插入最小占位 text block 以保持消息结构合法性。 - new_content = [{"type": "text", "text": "[thinking]"}] - logger.info( - "anthropic: inserted placeholder text block after stripping " - "%d thinking block(s) to avoid empty assistant content", - removed, - ) - message["content"] = new_content - stripped += removed - return stripped diff --git a/src/coding/proxy/server/routes.py b/src/coding/proxy/server/routes.py index d105ebe..5ad312d 100644 --- a/src/coding/proxy/server/routes.py +++ b/src/coding/proxy/server/routes.py @@ -24,14 +24,10 @@ logger = logging.getLogger(__name__) -async def _stream_proxy( - router: Any, body: dict, headers: dict, normalization: Any = None -) -> Any: +async def _stream_proxy(router: Any, body: dict, headers: dict) -> Any: """流式代理生成器.""" try: - async for chunk, vendor_name in router.route_stream( - body, headers, normalization=normalization - ): + async for chunk, vendor_name in router.route_stream(body, headers): yield chunk except NoCompatibleVendorError as exc: yield ( @@ -82,15 +78,13 @@ async def messages(request: Request) -> Response: if is_streaming: return StreamingResponse( - _stream_proxy(router, body, headers, normalization=normalization), + _stream_proxy(router, body, headers), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, ) try: - resp = await router.route_message( - body, headers, normalization=normalization - ) + resp = await router.route_message(body, headers) except NoCompatibleVendorError as exc: return json_error_response( 400, @@ -158,7 +152,7 @@ async def count_tokens(request: Request) -> Response: # count_tokens 无会话上下文，无法判断 thinking block 来源， # 安全剥离以防止跨供应商 signature 导致 400 错误。 - from .request_normalizer import strip_thinking_blocks + from ..convert.vendor_channels import strip_thinking_blocks strip_thinking_blocks(body) diff --git a/tests/test_request_normalizer.py b/tests/test_request_normalizer.py index 98ea238..bf653a9 100644 --- a/tests/test_request_normalizer.py +++ b/tests/test_request_normalizer.py @@ -4,23 +4,11 @@ import copy -from coding.proxy.server.request_normalizer import ( - apply_anthropic_specific_fixes, +from coding.proxy.convert.vendor_channels import ( enforce_anthropic_tool_pairing, - normalize_anthropic_request, strip_thinking_blocks, ) - - -def _apply_phase2(result): - """在 deep copy 上执行 Phase 2（Anthropic 专属修复），返回 (body, fixes).""" - body_copy = copy.deepcopy(result.body) - fixes = apply_anthropic_specific_fixes( - body_copy.get("messages", []), - result.misplaced_tool_results, - result.misplaced_log_info, - ) - return body_copy, fixes +from coding.proxy.server.request_normalizer import normalize_anthropic_request def test_rewrites_server_tool_use_to_standard_tool_use(): @@ -113,10 +101,10 @@ def test_unknown_tool_result_id_marks_fatal_reason(): class TestPhase1OnlyNormalization: - """验证 Phase 1（vendor-agnostic）不执行 Anthropic 专属修复. + """验证 Phase 1（vendor-agnostic）仅执行 ID 重写和 vendor block 移除. - Phase 1 仅执行 ID 重写、vendor block 移除、misplaced 信息收集。 - 重定位和孤儿修复由 Phase 2（apply_anthropic_specific_fixes）延迟执行。 + 跨供应商的 tool_use/tool_result 配对修复由 vendor_channels.py 中的 + enforce_anthropic_tool_pairing() 在源→目标转换通道中处理。 """ def test_phase1_keeps_misplaced_tool_result_in_place(self): @@ -147,12 +135,10 @@ def test_phase1_keeps_misplaced_tool_result_in_place(self): assert result.recoverable is True assert "misplaced_tool_result_relocated" not in result.adaptations assert "orphaned_tool_use_repaired" not in result.adaptations - # misplaced block 应仍在 assistant 消息中 + # misplaced block 应仍在 assistant 消息中（Phase 1 不处理） assistant_content = result.body["messages"][0]["content"] assert len(assistant_content) == 2 assert assistant_content[1]["type"] == "tool_result" - # Phase 2 上下文已收集 - assert len(result.misplaced_tool_results) == 1 def test_phase1_does_not_repair_orphans(self): """Phase 1 不应为孤儿 tool_use 合成 tool_result.""" @@ -178,467 +164,12 @@ def test_phase1_does_not_repair_orphans(self): assert "orphaned_tool_use_repaired" not in result.adaptations assert len(result.body["messages"]) == 1 # 无合成 user 消息 - -# ── 跨供应商 tool_result 位置错位重定位测试（Phase 1 + Phase 2）── - - -class TestMisplacedToolResultRelocation: - """验证 Phase 1 + Phase 2 组合能正确重定位 misplaced tool_result. - - Phase 1 收集 misplaced 信息，Phase 2 执行实际重定位。 - 此类测试模拟 executor 在 Anthropic tier 时的完整行为。 - """ - - def test_relocates_tool_result_from_assistant_message(self): - """assistant 消息中的 tool_result 经 Phase 2 重定位到新建的 user 消息.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "user", - "content": "run ls", - }, - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_123", - "name": "Bash", - "input": {"command": "ls"}, - }, - { - "type": "tool_result", - "tool_use_id": "toolu_123", - "content": "file1.txt\nfile2.txt", - }, - ], - }, - ], - } - ) - - # Phase 2 - body, fixes = _apply_phase2(result) - assert "misplaced_tool_result_relocated" in fixes - - messages = body["messages"] - assert len(messages) == 3 - # assistant 消息应只保留 tool_use - assistant_content = messages[1]["content"] - assert len(assistant_content) == 1 - assert assistant_content[0]["type"] == "tool_use" - assert assistant_content[0]["id"] == "toolu_123" - # 新增一条 user 消息包含被重定位的 tool_result - assert messages[2]["role"] == "user" - relocated_block = messages[2]["content"][0] - assert relocated_block["type"] == "tool_result" - assert relocated_block["tool_use_id"] == "toolu_123" - assert relocated_block["content"] == "file1.txt\nfile2.txt" - - def test_relocates_tool_result_preserves_other_blocks(self): - """重定位 tool_result 时保留同消息中的其他内容块.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - {"type": "text", "text": "Let me check."}, - { - "type": "tool_use", - "id": "toolu_456", - "name": "Read", - "input": {"path": "/etc/hosts"}, - }, - { - "type": "tool_result", - "tool_use_id": "toolu_456", - "content": "127.0.0.1 localhost", - }, - {"type": "text", "text": "Done."}, - ], - }, - ], - } - ) - - # Phase 2 - body, fixes = _apply_phase2(result) - assert "misplaced_tool_result_relocated" in fixes - - messages = body["messages"] - # assistant 消息保留 text + tool_use + text - assistant_content = messages[0]["content"] - assert len(assistant_content) == 3 - types = [b["type"] for b in assistant_content] - assert types == ["text", "tool_use", "text"] - # 新增 user 消息包含被重定位的 tool_result - assert len(messages) == 2 - assert messages[1]["role"] == "user" - assert messages[1]["content"][0]["type"] == "tool_result" - assert messages[1]["content"][0]["tool_use_id"] == "toolu_456" - - def test_tool_result_in_user_message_untouched(self): - """user 消息中的 tool_result 不受影响.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_789", - "name": "Bash", - "input": {"command": "echo hi"}, - }, - ], - }, - { - "role": "user", - "content": [ - { - "type": "tool_result", - "tool_use_id": "toolu_789", - "content": "hi", - }, - ], - }, - ], - } - ) - - assert result.recoverable is True - assert not result.misplaced_tool_results # 无 misplaced 块 - user_content = result.body["messages"][1]["content"] - assert len(user_content) == 1 - assert user_content[0]["type"] == "tool_result" - assert "misplaced_tool_result_relocated" not in result.adaptations - - def test_mixed_scenario_relocates_to_existing_user_message(self): - """assistant 和 user 消息中同时有 tool_result 时，assistant 中的被重定位到 user 消息.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_100", - "name": "Bash", - "input": {"command": "ls"}, - }, - { - "type": "tool_result", - "tool_use_id": "toolu_100", - "content": "misplaced result", - }, - ], - }, - { - "role": "user", - "content": [ - { - "type": "tool_result", - "tool_use_id": "toolu_100", - "content": "correct result", - }, - ], - }, - ], - } - ) - - # Phase 2 - body, fixes = _apply_phase2(result) - assert "misplaced_tool_result_relocated" in fixes - - # assistant 中的 tool_result 被移除 - assistant_content = body["messages"][0]["content"] - assert len(assistant_content) == 1 - assert assistant_content[0]["type"] == "tool_use" - # user 消息现在包含两个 tool_result（原有的 + 重定位来的） - user_content = body["messages"][1]["content"] - assert len(user_content) == 2 - assert all(b["type"] == "tool_result" for b in user_content) - - def test_deep_conversation_with_misplaced_tool_result_at_index_105(self): - """模拟长对话（105+ 消息）中 tool_result 出现在 assistant 消息的场景.""" - # 构建一个 106 条消息的对话历史 - messages = [] - for i in range(104): - role = "user" if i % 2 == 0 else "assistant" - messages.append( - { - "role": role, - "content": f"message {i}", - } - ) - - # 在消息 104（assistant）中放置 tool_use + tool_result - messages.append( - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_deep_1", - "name": "Bash", - "input": {"command": "find / -name '*.log'"}, - }, - { - "type": "tool_result", - "tool_use_id": "toolu_deep_1", - "content": "/var/log/system.log", - }, - ], - } - ) - # 消息 105（user） - messages.append( - { - "role": "user", - "content": "thanks", - } - ) - - result = normalize_anthropic_request({"messages": messages}) - - # Phase 2 - body, fixes = _apply_phase2(result) - assert "misplaced_tool_result_relocated" in fixes - - # 消息 104 中的 tool_result 应被移除 - assert len(body["messages"][104]["content"]) == 1 - assert body["messages"][104]["content"][0]["type"] == "tool_use" - # 消息 105（user）应包含原始文本 + 被重定位的 tool_result - user_content = body["messages"][105]["content"] - assert isinstance(user_content, list) - tool_result_blocks = [b for b in user_content if b.get("type") == "tool_result"] - text_blocks = [b for b in user_content if b.get("type") == "text"] - assert len(text_blocks) == 1 - assert text_blocks[0]["text"] == "thanks" - assert len(tool_result_blocks) == 1 - assert tool_result_blocks[0]["tool_use_id"] == "toolu_deep_1" - assert tool_result_blocks[0]["content"] == "/var/log/system.log" - - def test_rewrites_srvtoolu_id_when_relocating(self): - """重定位时同时重写 srvtoolu_ 前缀的 tool_use_id.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "server_tool_use", - "id": "srvtoolu_bad_1", - "name": "bash", - "input": {"cmd": "pwd"}, - }, - { - "type": "tool_result", - "tool_use_id": "srvtoolu_bad_1", - "content": "/home/user", - }, - ], - }, - ], - } - ) - - # Phase 2 - body, fixes = _apply_phase2(result) - assert "misplaced_tool_result_relocated" in fixes - assert "tool_result_tool_use_id_rewritten" in result.adaptations - - messages = body["messages"] - # assistant 中的 tool_use ID 已重写 - new_id = messages[0]["content"][0]["id"] - assert new_id.startswith("toolu_normalized_") - # 重定位到新 user 消息的 tool_result 使用相同的重写 ID - relocated = messages[1]["content"][0] - assert relocated["type"] == "tool_result" - assert relocated["tool_use_id"] == new_id - - -# ── 孤儿 tool_use 修复测试（Phase 1 + Phase 2）───────────────────── - - -class TestOrphanedToolUseRepair: - """验证 Phase 1 + Phase 2 组合能正确修复孤儿 tool_use. - - Phase 1 收集上下文（如 tool_id_map），Phase 2 执行孤儿修复。 - 此类测试模拟 executor 在 Anthropic tier 时的完整行为。 - """ - - def test_synthesizes_result_for_orphaned_tool_use(self): - """assistant 有 tool_use 且无后续 user 消息时，合成 user 消息含占位 tool_result.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_123", - "name": "Bash", - "input": {"command": "ls"}, - }, - ], - }, - ], - } - ) - - # Phase 2（手动调用，模拟 Anthropic tier 行为） - body, fixes = _apply_phase2(result) - assert "orphaned_tool_use_repaired" in fixes - - messages = body["messages"] - assert len(messages) == 2 - assert messages[1]["role"] == "user" - synthetic = messages[1]["content"][0] - assert synthetic["type"] == "tool_result" - assert synthetic["tool_use_id"] == "toolu_123" - assert synthetic["is_error"] is True - - def test_synthesizes_result_appends_to_existing_user(self): - """assistant 有 tool_use 且后续 user 消息为 text 时，追加合成 tool_result.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_456", - "name": "Bash", - "input": {"command": "pwd"}, - }, - ], - }, - { - "role": "user", - "content": "continue", - }, - ], - } - ) - - # Phase 2（手动调用，模拟 Anthropic tier 行为） - body, fixes = _apply_phase2(result) - assert "orphaned_tool_use_repaired" in fixes - - messages = body["messages"] - user_content = messages[1]["content"] - assert isinstance(user_content, list) - # 原始文本转为 text block + 合成的 tool_result - assert len(user_content) == 2 - assert user_content[0]["type"] == "text" - assert user_content[0]["text"] == "continue" - assert user_content[1]["type"] == "tool_result" - assert user_content[1]["tool_use_id"] == "toolu_456" - assert user_content[1]["is_error"] is True - - def test_synthesizes_only_missing_results(self): - """assistant 有 2 个 tool_use，user 仅有 1 个 tool_result 时，仅为缺失的合成.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_A", - "name": "Bash", - "input": {"command": "ls"}, - }, - { - "type": "tool_use", - "id": "toolu_B", - "name": "Read", - "input": {"path": "/etc/hosts"}, - }, - ], - }, - { - "role": "user", - "content": [ - { - "type": "tool_result", - "tool_use_id": "toolu_A", - "content": "file1.txt", - }, - ], - }, - ], - } - ) - - # Phase 2（手动调用，模拟 Anthropic tier 行为） - body, fixes = _apply_phase2(result) - assert "orphaned_tool_use_repaired" in fixes - - user_content = body["messages"][1]["content"] - assert len(user_content) == 2 - # 原有 tool_result 保持不变 - assert user_content[0]["tool_use_id"] == "toolu_A" - # 合成的 tool_result 仅针对 toolu_B - assert user_content[1]["type"] == "tool_result" - assert user_content[1]["tool_use_id"] == "toolu_B" - assert user_content[1]["is_error"] is True - - def test_no_repair_when_all_results_present(self): - """正常配对场景不应触发修复.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_ok", - "name": "Bash", - "input": {"command": "echo hi"}, - }, - ], - }, - { - "role": "user", - "content": [ - { - "type": "tool_result", - "tool_use_id": "toolu_ok", - "content": "hi", - }, - ], - }, - ], - } - ) - - assert result.recoverable is True - assert not result.misplaced_tool_results - # Phase 2 不应有任何修复 - body, fixes = _apply_phase2(result) - assert "orphaned_tool_use_repaired" not in fixes - - def test_repair_with_normalized_ids(self): - """跨供应商降级场景：srvtoolu_ ID 被重写后仍需修复孤儿 tool_use. - - 此场景下 tool_id_map 有条目，has_anthropic_fixes 为 True， - executor 会自动触发 Phase 2。 - """ + def test_phase1_then_enforce_pairing(self): + """Phase 1 + enforce_anthropic_tool_pairing 端到端测试（模拟完整链路）.""" result = normalize_anthropic_request( { "messages": [ + {"role": "user", "content": "hello"}, { "role": "assistant", "content": [ @@ -654,194 +185,37 @@ def test_repair_with_normalized_ids(self): "name": "Read", "input": {"path": "/tmp"}, }, - ], - }, - { - "role": "user", - "content": "continue", - }, - ], - } - ) - - # has_anthropic_fixes 为 True（因 tool_id_map 有条目） - assert result.has_anthropic_fixes is True - - # Phase 2 - body, fixes = _apply_phase2(result) - assert "orphaned_tool_use_repaired" in fixes - - messages = body["messages"] - # assistant 中 tool_use ID 已重写 - new_ids = [b["id"] for b in messages[0]["content"]] - assert len(new_ids) == 2 - assert all(id_.startswith("toolu_normalized_") for id_ in new_ids) - # user 消息应包含原始文本 + 两个合成的 tool_result - user_content = messages[1]["content"] - text_blocks = [b for b in user_content if b.get("type") == "text"] - synthetic_results = [ - b - for b in user_content - if b.get("type") == "tool_result" and b.get("is_error") is True - ] - assert len(text_blocks) == 1 - assert len(synthetic_results) == 2 - assert {r["tool_use_id"] for r in synthetic_results} == set(new_ids) - - -# ── has_cross_vendor_signals 属性测试 ────────────────────────── - - -class TestHasCrossVendorSignals: - """验证 NormalizationResult.has_cross_vendor_signals 属性.""" - - def test_true_when_tool_id_map_nonempty(self): - """tool_id_map 非空时标识为跨供应商信号.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "server_tool_use", - "id": "srvtoolu_abc", - "name": "Bash", - "input": {"command": "ls"}, - }, - ], - }, - ], - } - ) - assert result.tool_id_map # 非空 - assert result.has_cross_vendor_signals is True - - def test_true_when_misplaced_tool_results_exist(self): - """misplaced_tool_results 非空时标识为跨供应商信号.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_123", - "name": "Bash", - "input": {"command": "ls"}, - }, { "type": "tool_result", - "tool_use_id": "toolu_123", - "content": "output", - }, - ], - }, - ], - } - ) - assert result.misplaced_tool_results # 非空 - assert result.has_cross_vendor_signals is True - - def test_true_when_vendor_block_removed(self): - """vendor block 被移除时标识为跨供应商信号.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - {"type": "text", "text": "hello"}, - { - "type": "server_tool_use_delta", - "partial_json": '{"cmd":"pwd"}', - }, - ], - }, - ], - } - ) - assert "vendor_block_removed:server_tool_use_delta" in result.adaptations - assert result.has_cross_vendor_signals is True - - def test_true_when_invalid_tool_id_rewritten(self): - """非标准 tool_use ID 被重写时标识为跨供应商信号.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "glm_custom_id_123", - "name": "Bash", - "input": {"command": "ls"}, + "tool_use_id": "srvtoolu_X", + "content": "zhipu result", }, ], }, + {"role": "user", "content": "continue"}, ], } ) - assert "invalid_tool_use_id_rewritten_for_anthropic" in result.adaptations - assert result.has_cross_vendor_signals is True + assert result.recoverable - def test_false_for_pure_anthropic_request(self): - """纯 Anthropic 请求不应标识为跨供应商信号.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "thinking", - "thinking": "let me think", - "signature": "anthropic-valid-sig", - }, - {"type": "text", "text": "response"}, - ], - }, - {"role": "user", "content": "follow up"}, - ], - } - ) - assert not result.tool_id_map - assert not result.misplaced_tool_results - assert not result.adaptations - assert result.has_cross_vendor_signals is False + # 在 deep copy 上执行 enforce_anthropic_tool_pairing（模拟 vendor channel） + body_copy = copy.deepcopy(result.body) + fixes = enforce_anthropic_tool_pairing(body_copy.get("messages", [])) + assert "misplaced_tool_result_relocated" in fixes + assert "orphaned_tool_use_repaired" in fixes - def test_false_for_standard_tool_use(self): - """标准 toolu_ ID 不应标识为跨供应商信号.""" - result = normalize_anthropic_request( - { - "messages": [ - { - "role": "assistant", - "content": [ - { - "type": "tool_use", - "id": "toolu_standard_abc", - "name": "Bash", - "input": {"command": "ls"}, - }, - ], - }, - { - "role": "user", - "content": [ - { - "type": "tool_result", - "tool_use_id": "toolu_standard_abc", - "content": "output", - }, - ], - }, - ], - } - ) - assert result.has_cross_vendor_signals is False + messages = body_copy["messages"] + # assistant 只保留 2 个 tool_use + assistant_content = messages[1]["content"] + assert len(assistant_content) == 2 + assert all(b["type"] == "tool_use" for b in assistant_content) + # user 消息包含 2 个 tool_result + user_content = messages[2]["content"] + result_blocks = [b for b in user_content if b.get("type") == "tool_result"] + assert len(result_blocks) == 2 + expected_ids = {b["id"] for b in assistant_content} + result_ids = {b["tool_use_id"] for b in result_blocks} + assert result_ids == expected_ids # ── strip_thinking_blocks 函数测试 ────────────────────────────── @@ -996,7 +370,6 @@ def _enforce_pairing(messages): class TestEnforceAnthropicToolPairing: """验证 enforce_anthropic_tool_pairing 单遍强制配对函数. - 此函数替代原有多步串联管线（apply_anthropic_specific_fixes）， 通过单次正向遍历完成 tool_result 剥离、重定位和孤儿合成。 """ @@ -1436,59 +809,3 @@ def test_next_message_is_assistant_inserts_user(self): assert messages[1]["role"] == "user" assert messages[1]["content"][0]["type"] == "tool_result" assert messages[2]["role"] == "assistant" - - def test_phase1_then_enforce_pairing(self): - """Phase 1 + enforce_anthropic_tool_pairing 端到端测试（模拟完整链路）.""" - result = normalize_anthropic_request( - { - "messages": [ - {"role": "user", "content": "hello"}, - { - "role": "assistant", - "content": [ - { - "type": "server_tool_use", - "id": "srvtoolu_X", - "name": "Bash", - "input": {"command": "ls"}, - }, - { - "type": "server_tool_use", - "id": "srvtoolu_Y", - "name": "Read", - "input": {"path": "/tmp"}, - }, - { - "type": "tool_result", - "tool_use_id": "srvtoolu_X", - "content": "zhipu result", - }, - ], - }, - {"role": "user", "content": "continue"}, - ], - } - ) - assert result.recoverable - # Phase 1 重写了 ID，收集了 misplaced - assert result.tool_id_map - assert result.misplaced_tool_results - - # 在 deep copy 上执行 enforce_anthropic_tool_pairing（模拟 executor） - body_copy = copy.deepcopy(result.body) - fixes = enforce_anthropic_tool_pairing(body_copy.get("messages", [])) - assert "misplaced_tool_result_relocated" in fixes - assert "orphaned_tool_use_repaired" in fixes - - messages = body_copy["messages"] - # assistant 只保留 2 个 tool_use - assistant_content = messages[1]["content"] - assert len(assistant_content) == 2 - assert all(b["type"] == "tool_use" for b in assistant_content) - # user 消息包含 2 个 tool_result - user_content = messages[2]["content"] - result_blocks = [b for b in user_content if b.get("type") == "tool_result"] - assert len(result_blocks) == 2 - expected_ids = {b["id"] for b in assistant_content} - result_ids = {b["tool_use_id"] for b in result_blocks} - assert result_ids == expected_ids diff --git a/tests/test_vendor_channels.py b/tests/test_vendor_channels.py index ceb9f3b..7de4eae 100644 --- a/tests/test_vendor_channels.py +++ b/tests/test_vendor_channels.py @@ -4,7 +4,7 @@ - zhipu → anthropic 转换 (prepare_zhipu_to_anthropic) - zhipu → copilot 转换 (prepare_zhipu_to_copilot) - copilot → zhipu 转换 (prepare_copilot_to_zhipu) -- 共享辅助函数 (_strip_thinking_blocks_inplace, _strip_cache_control) +- 共享辅助函数 (strip_thinking_blocks, _strip_cache_control) - 转换注册表 (VENDOR_TRANSITIONS, get_transition_channel) """ @@ -15,18 +15,18 @@ from coding.proxy.convert.vendor_channels import ( VENDOR_TRANSITIONS, _strip_cache_control, - _strip_thinking_blocks_inplace, get_transition_channel, prepare_copilot_to_zhipu, prepare_zhipu_to_anthropic, prepare_zhipu_to_copilot, + strip_thinking_blocks, ) # ── 辅助函数测试 ────────────────────────────────────────────── -class TestStripThinkingBlocksInplace: - """_strip_thinking_blocks_inplace 单元测试.""" +class TestStripThinkingBlocks: + """strip_thinking_blocks 单元测试.""" def test_strips_thinking_blocks(self): body = { @@ -40,7 +40,7 @@ def test_strips_thinking_blocks(self): }, ] } - stripped = _strip_thinking_blocks_inplace(body) + stripped = strip_thinking_blocks(body) assert stripped == 1 assert body["messages"][0]["content"] == [ {"type": "text", "text": "response"}, @@ -57,7 +57,7 @@ def test_strips_redacted_thinking_blocks(self): }, ] } - stripped = _strip_thinking_blocks_inplace(body) + stripped = strip_thinking_blocks(body) assert stripped == 1 # content 为空时插入占位 text block assert body["messages"][0]["content"] == [ @@ -80,7 +80,7 @@ def test_inserts_placeholder_when_all_thinking(self): }, ] } - stripped = _strip_thinking_blocks_inplace(body) + stripped = strip_thinking_blocks(body) assert stripped == 2 assert body["messages"][1]["content"] == [ {"type": "text", "text": "[thinking]"}, @@ -95,7 +95,7 @@ def test_no_change_when_no_thinking(self): }, ] } - stripped = _strip_thinking_blocks_inplace(body) + stripped = strip_thinking_blocks(body) assert stripped == 0 assert body["messages"][0]["content"] == [{"type": "text", "text": "hello"}] @@ -105,7 +105,7 @@ def test_skips_non_assistant_messages(self): {"role": "user", "content": [{"type": "thinking", "thinking": "t"}]}, ] } - stripped = _strip_thinking_blocks_inplace(body) + stripped = strip_thinking_blocks(body) assert stripped == 0 def test_handles_string_content(self): @@ -114,7 +114,7 @@ def test_handles_string_content(self): {"role": "assistant", "content": "plain text"}, ] } - stripped = _strip_thinking_blocks_inplace(body) + stripped = strip_thinking_blocks(body) assert stripped == 0