Spaces:
Sleeping
Sleeping
| 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 |