Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-llm/src/lexigram/ai/llm/clients/_tools_utils.py: 18%

1"""Shared tool-calling helpers for LLM clients. 

2 

3Utilities to convert tool descriptors (``ToolDefinition``, classes 

4exposing ``__tool_schema__``, or OpenAI-format dicts) into provider wire 

5formats, parse provider ``tool_calls`` responses back into framework 

6:class:`ToolCall` objects, and serialize assistant tool-call / tool-result 

7messages for multi-turn round trips. 

8 

9The OpenAI-compatible helpers are reused by every provider whose API follows 

10the OpenAI chat-completions shape (OpenAI, Azure, Groq, Mistral, OpenRouter, 

11Cloudflare Workers AI, and the OpenAI-compatible family). 

12""" 

13 

14from __future__ import annotations 

15 

16from typing import Any 

17 

18from lexigram.ai.llm.clients._message_utils import serialize_content_for_openai 

19from lexigram.ai.llm.types import ChatMessage, FunctionCall, ToolCall 

20from lexigram.serialization import dumps_str, loads_str 

21 

22 

23def _tool_schema_fields(tool: Any) -> tuple[str | None, str, dict[str, Any]]: 

24 """Extract ``(name, description, parameters)`` from a tool descriptor. 

25 

26 Supports objects exposing a ``__tool_schema__`` class attribute (the 

27 Lexigram tool registration convention), ``ToolDefinition`` duck types 

28 (``name``/``description``/``parameters`` attributes), and OpenAI-format 

29 dicts (``{"function": {"name", "description", "parameters"}}``). 

30 

31 Args: 

32 tool: Any tool descriptor. 

33 

34 Returns: 

35 Tuple of (name, description, parameters). ``name`` is ``None`` when 

36 the descriptor has no usable name. 

37 """ 

38 schema = getattr(tool, "__tool_schema__", None) 

39 if isinstance(schema, dict) and schema.get("name"): 

40 return ( 

41 str(schema["name"]), 

42 str(schema.get("description", "") or ""), 

43 dict(schema.get("parameters", {}) or {}), 

44 ) 

45 if isinstance(tool, dict): 

46 func = tool.get("function", tool) 

47 if isinstance(func, dict): 

48 name = func.get("name") 

49 return ( 

50 str(name) if name else None, 

51 str(func.get("description", "") or ""), 

52 dict(func.get("parameters", {}) or {}), 

53 ) 

54 name = getattr(tool, "name", None) 

55 return ( 

56 str(name) if name else None, 

57 str(getattr(tool, "description", "") or ""), 

58 dict(getattr(tool, "parameters", {}) or {}), 

59 ) 

60 

61 

62def tool_to_openai_format(tool: Any) -> dict[str, Any] | None: 

63 """Convert a tool descriptor to OpenAI ``tools`` wire format. 

64 

65 Args: 

66 tool: Tool descriptor (``ToolDefinition``, schema class, or dict). 

67 

68 Returns: 

69 OpenAI tool dict (``{"type": "function", "function": {...}}``), or 

70 ``None`` when the descriptor has no usable name. 

71 """ 

72 name, description, parameters = _tool_schema_fields(tool) 

73 if not name: 

74 return None 

75 return { 

76 "type": "function", 

77 "function": { 

78 "name": name, 

79 "description": description, 

80 "parameters": parameters, 

81 }, 

82 } 

83 

84 

85def serialize_openai_tool_calls( 

86 tool_calls: list[ToolCall] | None, 

87) -> list[dict[str, Any]] | None: 

88 """Serialize framework ``ToolCall``s to OpenAI ``tool_calls`` wire format. 

89 

90 Used to re-emit an assistant turn that requested tools before the 

91 matching ``tool`` role responses. 

92 

93 Args: 

94 tool_calls: Framework tool calls from a prior assistant turn. 

95 

96 Returns: 

97 OpenAI ``tool_calls`` dicts, or ``None`` when empty. 

98 """ 

99 if not tool_calls: 

100 return None 

101 serialized: list[dict[str, Any]] = [] 

102 for call in tool_calls: 

103 if call.function is None: 

104 continue 

105 arguments = call.function.arguments 

106 serialized.append( 

107 { 

108 "id": call.id, 

109 "type": call.type or "function", 

110 "function": { 

111 "name": call.function.name, 

112 "arguments": ( 

113 arguments 

114 if isinstance(arguments, str) 

115 else dumps_str(arguments) 

116 ), 

117 }, 

118 } 

119 ) 

120 return serialized or None 

121 

122 

123def serialize_message_for_openai(msg: ChatMessage) -> dict[str, Any]: 

124 """Convert a ``ChatMessage`` to OpenAI message wire format. 

125 

126 Includes ``name``, ``tool_call_id`` (tool results), and ``tool_calls`` 

127 (assistant turns that requested tools) so multi-turn tool conversations 

128 round-trip correctly. 

129 

130 Args: 

131 msg: Chat message to convert. 

132 

133 Returns: 

134 OpenAI message dict. 

135 """ 

136 result: dict[str, Any] = { 

137 "role": msg.role.value if hasattr(msg.role, "value") else msg.role, 

138 "content": serialize_content_for_openai(msg.content), 

139 } 

140 if msg.name: 

141 result["name"] = msg.name 

142 if msg.tool_call_id: 

143 result["tool_call_id"] = msg.tool_call_id 

144 tool_calls = serialize_openai_tool_calls(msg.tool_calls) 

145 if tool_calls: 

146 if not msg.content: 

147 result["content"] = None 

148 result["tool_calls"] = tool_calls 

149 return result 

150 

151 

152def _attr_or_key(obj: Any, key: str) -> Any: 

153 """Read ``key`` from a dict or via attribute access (SDK objects).""" 

154 if isinstance(obj, dict): 

155 return obj.get(key) 

156 return getattr(obj, key, None) 

157 

158 

159def parse_openai_tool_calls(raw: Any) -> list[ToolCall] | None: 

160 """Parse OpenAI-format ``tool_calls`` into framework ``ToolCall`` objects. 

161 

162 Accepts the OpenAI SDK response objects (``choice.message.tool_calls``) 

163 as well as plain dicts from OpenAI-compatible REST providers. 

164 

165 Args: 

166 raw: List of tool call dicts or SDK objects (``None`` when absent). 

167 

168 Returns: 

169 List of :class:`ToolCall` objects, or ``None`` when empty. 

170 """ 

171 if not raw: 

172 return None 

173 calls: list[ToolCall] = [] 

174 for call in raw: 

175 fn = _attr_or_key(call, "function") or {} 

176 calls.append( 

177 ToolCall( 

178 id=str(_attr_or_key(call, "id") or ""), 

179 type=str(_attr_or_key(call, "type") or "function"), 

180 function=FunctionCall( 

181 name=str(_attr_or_key(fn, "name") or ""), 

182 arguments=_attr_or_key(fn, "arguments") or "", 

183 ), 

184 ) 

185 ) 

186 return calls or None 

187 

188 

189def parse_json_arguments(arguments: Any) -> dict[str, Any]: 

190 """Parse ``function.arguments`` (string or dict) into a plain dict. 

191 

192 Args: 

193 arguments: JSON-encoded string or dict from a provider response. 

194 

195 Returns: 

196 Plain argument dict (``{}`` when unparseable). 

197 """ 

198 if isinstance(arguments, dict): 

199 return dict(arguments) 

200 if isinstance(arguments, str): 

201 try: 

202 parsed = loads_str(arguments) 

203 except (TypeError, ValueError): 

204 return {} 

205 return dict(parsed) if isinstance(parsed, dict) else {} 

206 return {}