Spaces:
Sleeping
Sleeping
File size: 3,874 Bytes
c1ae554 | 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 | from __future__ import annotations
from typing import Dict, List, Optional
from .models import ChatMessage
from .helpers import normalize_content_to_list, segments_to_text
def reorder_messages_for_anthropic(history: List[ChatMessage]) -> List[ChatMessage]:
if not history:
return []
expanded: List[ChatMessage] = []
for m in history:
if m.role == "user":
items = normalize_content_to_list(m.content)
if isinstance(m.content, list) and len(items) > 1:
for seg in items:
if isinstance(seg, dict) and seg.get("type") == "text" and isinstance(seg.get("text"), str):
expanded.append(ChatMessage(role="user", content=seg.get("text")))
else:
expanded.append(ChatMessage(role="user", content=[seg] if isinstance(seg, dict) else seg))
else:
expanded.append(m)
elif m.role == "assistant" and m.tool_calls and len(m.tool_calls) > 1:
_assistant_text = segments_to_text(normalize_content_to_list(m.content))
if _assistant_text:
expanded.append(ChatMessage(role="assistant", content=_assistant_text))
for tc in (m.tool_calls or []):
expanded.append(ChatMessage(role="assistant", content=None, tool_calls=[tc]))
else:
expanded.append(m)
last_input_tool_id: Optional[str] = None
last_input_is_tool = False
for m in reversed(expanded):
if m.role == "tool" and m.tool_call_id:
last_input_tool_id = m.tool_call_id
last_input_is_tool = True
break
if m.role == "user":
break
tool_results_by_id: Dict[str, ChatMessage] = {}
assistant_tc_ids: set[str] = set()
for m in expanded:
if m.role == "tool" and m.tool_call_id and m.tool_call_id not in tool_results_by_id:
tool_results_by_id[m.tool_call_id] = m
if m.role == "assistant" and m.tool_calls:
try:
for tc in (m.tool_calls or []):
_id = (tc or {}).get("id")
if isinstance(_id, str) and _id:
assistant_tc_ids.add(_id)
except Exception:
pass
result: List[ChatMessage] = []
trailing_assistant_msg: Optional[ChatMessage] = None
for m in expanded:
if m.role == "tool":
# Preserve unmatched tool results inline
if not m.tool_call_id or m.tool_call_id not in assistant_tc_ids:
result.append(m)
if m.tool_call_id:
tool_results_by_id.pop(m.tool_call_id, None)
continue
if m.role == "assistant" and m.tool_calls:
ids: List[str] = []
try:
for tc in (m.tool_calls or []):
_id = (tc or {}).get("id")
if isinstance(_id, str) and _id:
ids.append(_id)
except Exception:
pass
if last_input_is_tool and last_input_tool_id and (last_input_tool_id in ids):
if trailing_assistant_msg is None:
trailing_assistant_msg = m
continue
result.append(m)
for _id in ids:
tr = tool_results_by_id.pop(_id, None)
if tr is not None:
result.append(tr)
continue
result.append(m)
if last_input_is_tool and last_input_tool_id and trailing_assistant_msg is not None:
result.append(trailing_assistant_msg)
tr = tool_results_by_id.pop(last_input_tool_id, None)
if tr is not None:
result.append(tr)
return result |