-
Notifications
You must be signed in to change notification settings - Fork 2
Expand file tree
/
Copy pathvendor_channels.py
More file actions
841 lines (691 loc) · 32.9 KB
/
Copy pathvendor_channels.py
File metadata and controls
841 lines (691 loc) · 32.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
"""供应商跨供应商转换通道 — 源→目标绑定的请求体预处理.
每个通道函数表示一个具体的「源 vendor → 目标 vendor」绑定转换关系,
接收标准化的 Anthropic 格式请求体,返回清理跨供应商产物后的请求体(深拷贝)。
通道注册表 ``VENDOR_TRANSITIONS`` 提供统一的 (source, target) → channel_fn 映射,
executor 层通过 ``get_transition_channel()`` 查表分发,无需感知具体供应商逻辑。
转换矩阵(仅注册需要转换的源→目标对,未注册的不触发任何通道):
zhipu → anthropic : prepare_zhipu_to_anthropic (剥离 thinking + tool pairing)
zhipu → copilot : prepare_zhipu_to_copilot (剥离 thinking + cache_control + tool pairing)
copilot → zhipu : prepare_copilot_to_zhipu (剥离 thinking + cache_control + 移除 thinking 参数 + tool pairing)
"""
from __future__ import annotations
import copy
import logging
import re
from collections.abc import Callable
from typing import Any
logger = logging.getLogger(__name__)
_THINKING_BLOCK_TYPES = {"thinking", "redacted_thinking"}
# ── Anthropic 工具块 ID 规范 ───────────────────────────────────
_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_]+$")
# Zhipu 流式响应中出现的非标准供应商私有 content block 类型.
# Anthropic API 拒绝这些块,需要在跨 vendor 请求体中剥离.
_ZHIPU_VENDOR_BLOCK_TYPES = {"server_tool_use_delta"}
# Zhipu 内联输出非标准 content block 类型的标识(用于源供应商推断).
_ZHIPU_SERVER_TOOL_USE_TYPES = {"server_tool_use", "server_tool_use_delta"}
# ── 转换通道注册表 ─────────────────────────────────────────────
# (source_vendor, target_vendor) → (body) → (prepared_body, adaptations)
VENDOR_TRANSITIONS: dict[
tuple[str, str], Callable[[dict[str, Any]], tuple[dict[str, Any], list[str]]]
] = {}
def get_transition_channel(
source: str, target: str
) -> Callable[[dict[str, Any]], tuple[dict[str, Any], list[str]]] | None:
"""查找源→目标绑定转换通道,不存在时返回 None."""
return VENDOR_TRANSITIONS.get((source, target))
# ── 共享辅助函数 ──────────────────────────────────────────────
def _dump_message_digest(
messages: list[dict[str, Any]],
*,
max_messages: int = 10,
label: str = "",
) -> None:
"""输出前 N 条消息的结构摘要(role + content_type_counts),用于过渡管线诊断.
仅在 DEBUG 级别输出,且仅在消息数 > 0 时才输出,避免噪声。
"""
if not messages or not logger.isEnabledFor(logging.DEBUG):
return
parts: list[str] = [f"[{label}]" if label else ""]
limit = min(len(messages), max_messages)
for idx in range(limit):
msg = messages[idx]
role = msg.get("role", "?") if isinstance(msg, dict) else "?"
content = msg.get("content") if isinstance(msg, dict) else None
if isinstance(content, list):
type_counts: dict[str, int] = {}
for b in content:
if isinstance(b, dict):
t = b.get("type", "?")
type_counts[t] = type_counts.get(t, 0) + 1
else:
type_counts["raw"] = type_counts.get("raw", 0) + 1
counts_str = ",".join(f"{t}:{c}" for t, c in sorted(type_counts.items()))
elif isinstance(content, str):
counts_str = f"str({len(content)})"
else:
counts_str = "empty"
parts.append(f"{idx}:{role}[{counts_str}]")
if len(messages) > max_messages:
parts.append(f"...+{len(messages) - max_messages}more")
logger.debug("Transition digest %s", " ".join(parts))
def _validate_anthropic_pairing(
messages: list[dict[str, Any]],
*,
context: str = "",
) -> list[str]:
"""独立的 Anthropic tool_use/tool_result 配对自检(过渡管线末端执行).
与 ``_enforce_pairing_sanity_pass`` 不同,此函数:
- 不修改消息列表(纯检测)
- 针对每个 assistant + tool_use,精确记录下一条 user 消息中匹配/缺失的 ID
- 发现不一致时输出 WARNING 级别日志含 message index 与具体 ID
Returns:
检测到的问题描述列表(空列表表示全部通过)。
"""
issues: list[str] = []
for i, msg in enumerate(messages):
if not isinstance(msg, dict) or msg.get("role") != "assistant":
continue
content = msg.get("content")
if not isinstance(content, list):
continue
tool_use_ids = [
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:
continue
next_idx = i + 1
if next_idx >= len(messages):
issues.append(f"messages[{i}]: assistant with tool_uses at end of list")
continue
next_msg = messages[next_idx]
if not isinstance(next_msg, dict) or next_msg.get("role") != "user":
issues.append(
f"messages[{i}]: next messages[{next_idx}] is not user "
f"(role={next_msg.get('role') if isinstance(next_msg, dict) else '?'})"
)
continue
user_content = next_msg.get("content")
if not isinstance(user_content, list):
user_content = []
result_ids = {
b["tool_use_id"]
for b in user_content
if isinstance(b, dict)
and b.get("type") == "tool_result"
and isinstance(b.get("tool_use_id"), str)
}
missing = [uid for uid in tool_use_ids if uid not in result_ids]
if missing:
issue = (
f"messages[{i}]: {len(missing)}/{len(tool_use_ids)} tool_use(s) "
f"without tool_result in messages[{next_idx}]: {missing[:5]}"
)
issues.append(issue)
if issues:
prefix = f"[{context}] " if context else ""
logger.warning(
"Anthropic pairing validation: %s%d issue(s) found: %s",
prefix,
len(issues),
"; ".join(issues),
)
return issues
def strip_thinking_blocks(body: dict[str, Any]) -> int:
"""从 assistant 消息中移除 thinking/redacted_thinking 块(就地).
Anthropic API 要求 thinking blocks 的 signature 必须是其签发的有效签名。
跨供应商迁移(如 Zhipu → Anthropic)后,conversation history 中可能包含
非 Anthropic 签发的 signature,导致 400 invalid_request_error。
根据 Anthropic 官方文档,thinking blocks 可以被安全省略,不影响模型行为。
剥离后 content 为空时插入最小占位 text block 以保持消息结构合法性。
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:
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),
)
# 纵深防御: sanity 兜底,捕获主循环未覆盖的边角配对漏洞
adaptations.extend(_enforce_pairing_sanity_pass(messages_list))
return adaptations
def _enforce_pairing_sanity_pass(
messages_list: list[dict[str, Any]],
) -> list[str]:
"""``enforce_anthropic_tool_pairing`` 主循环之后的纯检测兜底 helper.
职责正交于主循环(不剥离 tool_result、不插入新 user 消息),仅做两件事:
1. 遍历每个 ``role == "assistant"`` 且包含 ``tool_use`` 块的消息,
检查 ``messages[i+1]`` 是否为 ``user`` 且包含所有 ``tool_use.id`` 对应
``tool_result.tool_use_id``。
2. 缺失项在该 user 消息末尾追加 ``is_error=True`` 占位块;如果 next 消息根本
不是 user(主循环未触达此分支的退化场景),同样不做插入,仅记录 WARNING
供运维定位 —— 该路径正常情况下永不命中(主循环已保证 next user 存在)。
本 helper 单独抽出的目的有两个:
- 直接构造"绕过主循环"的输入做单元测试,确保 sanity 分支具备**正向回归保护**
(历史教训: ``9061cd0`` 引入两遍扫描+sanity 后被 ``2bac9a7`` 连带回滚,
重要原因之一是缺乏对兜底路径的独立单测)。
- 在主循环 A-F 步骤未来重构时,sanity 仍能稳定守住 Anthropic 配对约束。
Args:
messages_list: 消息列表(就地修改)。
Returns:
新增的 adaptation 标签列表(命中则为 ``["pairing_sanity_repaired"]``,否则空列表)。
"""
repaired: list[tuple[int, str]] = []
for i, msg in enumerate(messages_list):
if not isinstance(msg, dict) or msg.get("role") != "assistant":
continue
content = msg.get("content")
if not isinstance(content, list):
continue
tool_use_ids = [
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:
continue
next_idx = i + 1
if (
next_idx >= len(messages_list)
or not isinstance(messages_list[next_idx], dict)
or messages_list[next_idx].get("role") != "user"
):
# 主循环正常情况下已保证 next 为 user;此处仅日志告警,不做隐式插入
# 以避免与主循环职责重叠。
logger.warning(
"Sanity pass: assistant at messages[%d] has tool_use without "
"user next message (tool_use_ids=%s). Main enforce loop may have a regression.",
i,
", ".join(tool_use_ids),
)
continue
user_msg = messages_list[next_idx]
user_content = user_msg.get("content")
if not isinstance(user_content, list):
# 主循环 D 步已将 string content 归一化为 list;这里防御性兜底
user_msg["content"] = (
[{"type": "text", "text": user_content}]
if isinstance(user_content, str)
else []
)
user_content = user_msg["content"]
existing_result_ids = {
b["tool_use_id"]
for b in user_content
if isinstance(b, dict)
and b.get("type") == "tool_result"
and b.get("tool_use_id")
}
for uid in tool_use_ids:
if uid in existing_result_ids:
continue
user_content.append(
{
"type": "tool_result",
"tool_use_id": uid,
"content": "",
"is_error": True,
}
)
repaired.append((i, uid))
if not repaired:
return []
logger.warning(
"Sanity pass repaired %d unpaired tool_use(s) missed by main enforce loop. "
"Affected: %s",
len(repaired),
", ".join(f"messages[{idx}]:{uid}" for idx, uid in repaired),
)
return ["pairing_sanity_repaired"]
def _strip_cache_control(body: dict[str, Any]) -> int:
"""从 system/messages/tools 中移除 cache_control 字段(就地).
部分供应商(GLM-5、OpenAI)不支持 Anthropic 的 cache_control 扩展,
保留该字段可能导致请求被拒绝或产生意外行为。
Returns:
被移除的 cache_control 字段数量。
"""
removed = 0
# System prompt blocks
system = body.get("system")
if isinstance(system, list):
for block in system:
if isinstance(block, dict) and "cache_control" in block:
del block["cache_control"]
removed += 1
# Message content blocks
for message in body.get("messages", []):
if not isinstance(message, dict):
continue
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and "cache_control" in block:
del block["cache_control"]
removed += 1
# Tools
for tool in body.get("tools", []):
if isinstance(tool, dict) and "cache_control" in tool:
del tool["cache_control"]
removed += 1
return removed
# ── zhipu 共享清洗函数 ──────────────────────────────────────────
# 跨供应商转换时主动剥离的顶层参数。
# 首选 tier 场景的 thinking.type=adaptive 兼容转换由
# ZhipuVendor._prepare_request 处理(转换为 enabled + budget,保留功能),
# 此处仅负责 failover 路径的全量剥离(跨供应商 thinking signature 失效)。
_ZHIPU_UNSUPPORTED_PARAMS: frozenset[str] = frozenset(
{"thinking", "extended_thinking", "reasoning_effort"}
)
def normalize_for_zhipu(body: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
"""为 zhipu GLM 的 Anthropic 兼容端点清洗请求体(就地,不 deep copy).
为跨供应商转换通道 ``prepare_copilot_to_zhipu`` 提供请求体清洗。
清洗内容:
1. 剥离 cache_control 字段(GLM 静默忽略,主动剥离以减少噪音)
2. 移除顶层 thinking/extended_thinking/reasoning_effort 参数(GLM 原生支持
thinking、静默忽略 reasoning_effort,但跨供应商场景下这些参数来自原供应商
的协议语义,主动剥离以确保请求语义一致性)
3. 强制 tool_use/tool_result 配对约束
不包含 thinking blocks 剥离:跨供应商场景下 history 中的 thinking blocks
来自原供应商(签名失效),由调用方在调用本函数之前通过
``strip_thinking_blocks`` 单独处理。
所有操作均为幂等,安全地在已清洗的请求体上重复调用。
Returns:
(body, adaptations) — body 为就地修改后的同一引用,adaptations 为变换描述列表。
"""
adaptations: list[str] = []
# Step 1: 剥离 cache_control
removed_cc = _strip_cache_control(body)
if removed_cc:
adaptations.append(f"removed_{removed_cc}_cache_control_fields")
# Step 2: 移除不支持的顶层参数
for param in _ZHIPU_UNSUPPORTED_PARAMS:
if param in body:
del body[param]
adaptations.append(f"removed_{param}_param")
# Step 3: 强制 tool_use/tool_result 配对
pairing_fixes = enforce_anthropic_tool_pairing(body.get("messages", []))
if pairing_fixes:
adaptations.extend(pairing_fixes)
return body, adaptations
def _remove_vendor_blocks(body: dict[str, Any], block_types: set[str]) -> int:
"""从 messages[].content[] 中就地移除指定 type 的内容块.
用于剥离 vendor 私有 content block 类型(如 zhipu 的 ``server_tool_use_delta``),
Anthropic API 会拒绝这些非标准块。
Returns:
被移除的块数量。
"""
removed = 0
for message in body.get("messages", []):
if not isinstance(message, dict):
continue
content = message.get("content")
if not isinstance(content, list):
continue
new_content: list[Any] = []
for block in content:
if isinstance(block, dict) and block.get("type") in block_types:
removed += 1
continue
new_content.append(block)
if removed:
message["content"] = new_content
return removed
def _rewrite_srvtoolu_ids(body: dict[str, Any]) -> tuple[int, dict[str, str]]:
"""将 zhipu 的 ``server_tool_use`` + ``srvtoolu_*`` ID 改写为标准 Anthropic 形式.
Anthropic API 要求 tool_use 类型与 ``toolu_*`` 格式的 ID。Zhipu 的
``server_tool_use`` + ``srvtoolu_*`` 在上游 Anthropic 兼容端点可用,但无法
透传至其他供应商;同时还需重写所有 ``tool_result.tool_use_id`` 引用,保持配对关系。
**两遍扫描(消除块顺序敏感性)**:
- Pass 1: 仅遍历 ``role == "assistant"`` 的消息,按 assistant 出现顺序为每个
待改写的 tool_use 分配 ``toolu_normalized_N`` 新 ID,建立完整 ``id_map``。
- Pass 2: 全量遍历消息,对任意 ``tool_result.tool_use_id ∈ id_map`` 的块
原地改写为新 ID(不分 user / assistant,覆盖 misplaced 与跨消息边界场景)。
单遍方案在 GLM-5 偶发将 inline ``tool_result`` 输出在对应 ``server_tool_use``
之前的乱序场景下,会因 Case B 时 ``id_map`` 尚未填入而漏改 ``tool_use_id``,
导致 ``enforce_anthropic_tool_pairing`` 后 ``extracted_tool_results`` 的 key
与 ``tool_use_ids`` 不一致,进而把本应配对的 misplaced tool_result 默默丢弃,
最终触发 Anthropic ``messages.x: tool_use ids were found without tool_result
blocks immediately after`` 400 错误。两遍扫描以"先建表、后改写"的次序消除该
时序耦合。
Returns:
(rewritten_count, id_map) — 重写次数与 {原 ID: 新 ID} 映射。
"""
id_map: dict[str, str] = {}
counter = 0
def next_id() -> str:
nonlocal counter
counter += 1
return f"toolu_normalized_{counter}"
# Pass 1: 扫描 assistant 消息,改写 tool_use / server_tool_use 的 id 与 type,
# 按出现顺序填充 id_map(保持与单遍版本相同的序号分配,避免破坏既有断言)。
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
for block in content:
if not isinstance(block, dict):
continue
block_type = block.get("type")
if block_type not in {"tool_use", "server_tool_use"}:
continue
block_id = block.get("id")
if isinstance(block_id, str) and _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(
block_id
):
new_id = next_id()
id_map[block_id] = new_id
block["id"] = new_id
block["type"] = "tool_use"
elif (
isinstance(block_id, str)
and block_id
and not _ANTHROPIC_TOOL_USE_ID_RE.match(block_id)
and block.get("name")
):
# 非标准 ID(非 toolu_ / srvtoolu_),且具备 name 可改写
new_id = next_id()
id_map[block_id] = new_id
block["id"] = new_id
block["type"] = "tool_use"
elif block_type == "server_tool_use" and isinstance(block_id, str):
# 兜底: 类型是 server_tool_use 但 ID 已是标准 toolu_ 形式,仅纠正类型
block["type"] = "tool_use"
# Pass 2: 全量扫描,对任意 tool_result.tool_use_id 命中 id_map 的块同步改写。
if id_map:
for message in body.get("messages", []):
if not isinstance(message, dict):
continue
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if not isinstance(block, dict):
continue
if block.get("type") != "tool_result":
continue
tool_use_id = block.get("tool_use_id")
if isinstance(tool_use_id, str) and tool_use_id in id_map:
block["tool_use_id"] = id_map[tool_use_id]
return len(id_map), id_map
def infer_source_vendor_from_body(body: dict[str, Any]) -> str | None:
"""从请求 body 内容推断源供应商(仅在无会话上下文时作为兜底).
启发式(按置信度排序):
- 出现 ``srvtoolu_*`` 格式的 ``tool_use.id`` → zhipu
- 出现 ``server_tool_use`` / ``server_tool_use_delta`` 类型的 content block → zhipu
原则: 只读扫描不修改 body;无匹配返回 None(视作纯净无需跨供应商清洗)。
Args:
body: Anthropic Messages 请求体。
Returns:
推断的源供应商名称(当前仅支持 ``"zhipu"``),无法推断返回 None。
"""
for message in body.get("messages", []):
if not isinstance(message, dict):
continue
content = message.get("content")
if not isinstance(content, list):
continue
for block in content:
if not isinstance(block, dict):
continue
block_type = block.get("type")
if block_type in _ZHIPU_SERVER_TOOL_USE_TYPES:
return "zhipu"
block_id = block.get("id")
if isinstance(block_id, str) and _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(
block_id
):
return "zhipu"
tool_use_id = block.get("tool_use_id")
if isinstance(tool_use_id, str) and _ANTHROPIC_SERVER_TOOL_USE_ID_RE.match(
tool_use_id
):
return "zhipu"
return None
# ── copilot → zhipu 转换通道 ─────────────────────────────────────
def prepare_copilot_to_zhipu(
body: dict[str, Any],
) -> tuple[dict[str, Any], list[str]]:
"""copilot → zhipu 转换: 清理 copilot 产物以适配 GLM-5.
GLM-5 的 Anthropic 兼容端点对以下特性支持不完整:
- thinking / redacted_thinking 块 (signature 由非 Anthropic 签发)
- cache_control 字段
- 跨供应商产物 (misplaced tool_result, 非标准 tool_use ID)
- 顶层 thinking / extended_thinking 参数
Returns:
(prepared_body, adaptations) — adaptations 为应用的变换描述列表。
"""
prepared = copy.deepcopy(body)
adaptations: list[str] = []
# Step 1: 剥离 thinking/redacted_thinking 块(跨供应商签名失效)
stripped = strip_thinking_blocks(prepared)
if stripped:
adaptations.append(f"stripped_{stripped}_thinking_blocks")
# Step 2: 共享清洗(cache_control、不支持的顶层参数、tool pairing)
_, norm_adaptations = normalize_for_zhipu(prepared)
adaptations.extend(norm_adaptations)
return prepared, adaptations
# ── zhipu → copilot 转换通道 ─────────────────────────────────────
def prepare_zhipu_to_copilot(
body: dict[str, Any],
) -> tuple[dict[str, Any], list[str]]:
"""zhipu → copilot 转换: 清理 zhipu 产物以确保 OpenAI 转换器稳定.
Copilot 内部的 convert_openai_request() 处理 Anthropic→OpenAI 格式转换,
但 zhipu 的跨供应商产物可能导致转换器输出异常:
- 非 Anthropic 签发的 thinking signature
- cache_control 字段(OpenAI 协议不支持)
- 错位的 tool_result blocks
注意: 不移除顶层 thinking 参数,由 copilot converter 自行映射。
Returns:
(prepared_body, adaptations) — adaptations 为应用的变换描述列表。
"""
prepared = copy.deepcopy(body)
adaptations: list[str] = []
# Step 1: 剥离 zhipu 私有 content block 类型
removed_vendor_blocks = _remove_vendor_blocks(prepared, _ZHIPU_VENDOR_BLOCK_TYPES)
if removed_vendor_blocks:
adaptations.append(f"removed_{removed_vendor_blocks}_zhipu_vendor_blocks")
# Step 2: 改写 srvtoolu_* ID 与 server_tool_use 类型
rewritten, _ = _rewrite_srvtoolu_ids(prepared)
if rewritten:
adaptations.append(f"rewritten_{rewritten}_srvtoolu_ids")
# Step 3: 剥离 thinking/redacted_thinking 块
stripped = strip_thinking_blocks(prepared)
if stripped:
adaptations.append(f"stripped_{stripped}_thinking_blocks")
# Step 4: 移除 cache_control 字段
removed_cc = _strip_cache_control(prepared)
if removed_cc:
adaptations.append(f"removed_{removed_cc}_cache_control_fields")
# Step 5: 强制 tool_use/tool_result 配对
pairing_fixes = enforce_anthropic_tool_pairing(prepared.get("messages", []))
if pairing_fixes:
adaptations.extend(pairing_fixes)
return prepared, adaptations
# ── zhipu → anthropic 转换通道 ────────────────────────────────────
def prepare_zhipu_to_anthropic(
body: dict[str, Any],
) -> tuple[dict[str, Any], list[str]]:
"""zhipu → anthropic 转换: 清理 zhipu 产物以适配 Anthropic API.
Anthropic API 要求:
- tool_use 类型与 ``toolu_*`` 格式 ID(zhipu 的 ``server_tool_use``/``srvtoolu_*`` 不兼容)
- 每个 tool_use 必须在紧随的 user 消息中有对应 tool_result
- thinking blocks 的 signature 必须是 Anthropic 签发(zhipu 签发的无效)
- 不接受 ``server_tool_use_delta`` 等 zhipu 私有流式块类型
此通道按顺序执行:
1. 剥离 zhipu 私有 block 类型(``server_tool_use_delta``)
2. 改写 ``srvtoolu_*`` ID 与 ``server_tool_use`` 类型为标准 Anthropic 形式
3. 强制 tool_use/tool_result 配对(单遍正向扫描)
4. 剥离 thinking blocks(signature 无效)
5. 独立的 Anthropic 兼容性自检(纯检测,不修改,定位 enforce/sanity 未覆盖的边界 case)
所有变换均为幂等操作,安全地在已清理的请求体上重复执行。
Returns:
(prepared_body, adaptations) — adaptations 为应用的变换描述列表。
"""
prepared = copy.deepcopy(body)
adaptations: list[str] = []
msgs = prepared.get("messages", [])
# ── 过渡管线诊断:变换前快照 ──
_dump_message_digest(msgs, label="zhipu→anthropic.before")
# Step 1: 剥离 zhipu 私有 content block 类型(如 server_tool_use_delta)
removed_vendor_blocks = _remove_vendor_blocks(prepared, _ZHIPU_VENDOR_BLOCK_TYPES)
if removed_vendor_blocks:
adaptations.append(f"removed_{removed_vendor_blocks}_zhipu_vendor_blocks")
# Step 2: 改写 srvtoolu_* ID 与 server_tool_use 类型
rewritten, _ = _rewrite_srvtoolu_ids(prepared)
if rewritten:
adaptations.append(f"rewritten_{rewritten}_srvtoolu_ids")
_dump_message_digest(msgs, label="zhipu→anthropic.after_rewrite")
# Step 3: 强制 tool_use/tool_result 配对
pairing_fixes = enforce_anthropic_tool_pairing(msgs)
if pairing_fixes:
adaptations.extend(pairing_fixes)
_dump_message_digest(msgs, label="zhipu→anthropic.after_enforce")
# Step 4: 剥离 thinking blocks(zhipu signature 无效)
stripped = strip_thinking_blocks(prepared)
if stripped:
adaptations.append(f"stripped_{stripped}_thinking_blocks")
_dump_message_digest(msgs, label="zhipu→anthropic.after_strip")
# Step 5: 独立的 Anthropic 兼容性自检(纯检测,不修改)
validation_issues = _validate_anthropic_pairing(msgs, context="zhipu→anthropic")
if validation_issues:
adaptations.append("anthropic_pairing_validation_issues")
return prepared, adaptations
# ── 注册所有转换通道 ──────────────────────────────────────────────
VENDOR_TRANSITIONS[("zhipu", "anthropic")] = prepare_zhipu_to_anthropic
VENDOR_TRANSITIONS[("zhipu", "copilot")] = prepare_zhipu_to_copilot
VENDOR_TRANSITIONS[("copilot", "zhipu")] = prepare_copilot_to_zhipu