Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-relay/src/lexigram/ai/relay/mappers/openai_chat/_helpers.py: 30%

1"""Constants and free helpers shared across OpenAI Chat conversion.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.ai.relay.context import ConversionContext 

8from lexigram.ai.relay.mappers.base import record_loss 

9from lexigram.contracts.ai.llm import FunctionCall, ToolCall 

10from lexigram.contracts.ai.relay.types import RelayFormat 

11from lexigram.serialization import dumps_str 

12 

13_TARGET = RelayFormat.OPENAI_CHAT 

14_MESSAGE_METADATA_INTERNAL = {"function_call_item_ids"} 

15 

16 

17def _tool_calls_to_ir( 

18 wire: list[dict[str, Any]] | None, 

19) -> list[ToolCall] | None: 

20 """Convert wire tool-call dicts into canonical ``ToolCall`` objects.""" 

21 if not wire: 

22 return None 

23 tool_calls: list[ToolCall] = [] 

24 for item in wire: 

25 function = item.get("function") 

26 name = function.get("name", "") if isinstance(function, dict) else "" 

27 arguments = function.get("arguments", {}) if isinstance(function, dict) else {} 

28 tool_calls.append( 

29 ToolCall( 

30 id=str(item.get("id", "")), 

31 type=str(item.get("type", "function")), 

32 function=FunctionCall(name=str(name), arguments=arguments), 

33 ) 

34 ) 

35 return tool_calls 

36 

37 

38def _tool_call_to_wire(tool_call: ToolCall) -> dict[str, Any]: 

39 """Serialize one canonical ``ToolCall`` as a wire dict.""" 

40 arguments: Any = tool_call.function.arguments if tool_call.function else {} 

41 if isinstance(arguments, dict): 

42 arguments = dumps_str(arguments) 

43 elif not isinstance(arguments, str): 

44 arguments = "" 

45 return { 

46 "id": tool_call.id, 

47 "type": "function", 

48 "function": { 

49 "name": tool_call.function.name if tool_call.function else "", 

50 "arguments": arguments, 

51 }, 

52 } 

53 

54 

55def _extract_text( 

56 content: str | list[dict[str, Any]] | None, 

57 context: ConversionContext, 

58 *, 

59 field: str, 

60) -> str: 

61 """Extract the text portion of wire content for flattened fields.""" 

62 if content is None: 

63 return "" 

64 if isinstance(content, str): 

65 return content 

66 texts: list[str] = [] 

67 lost = False 

68 for part in content: 

69 if isinstance(part, dict) and part.get("type") == "text": 

70 texts.append(str(part.get("text", ""))) 

71 else: 

72 lost = True 

73 if lost: 

74 record_loss( 

75 context, field=field, target=_TARGET, reason="non_text_parts_dropped" 

76 ) 

77 return "".join(texts)