-
Notifications
You must be signed in to change notification settings - Fork 367
Expand file tree
/
Copy pathopenai_client.py
More file actions
295 lines (246 loc) · 9.91 KB
/
openai_client.py
File metadata and controls
295 lines (246 loc) · 9.91 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
"""OpenAI LLM client implementation."""
import json
import logging
from typing import Any, Optional, Union
from openai import AsyncOpenAI
from ..retry import RetryConfig, async_retry
from ..schema import FunctionCall, LLMResponse, Message, TokenUsage, ToolCall
from .base import LLMClientBase
logger = logging.getLogger(__name__)
class OpenAIClient(LLMClientBase):
"""LLM client using OpenAI's protocol.
This client uses the official OpenAI SDK and supports:
- Reasoning content (via reasoning_split=True)
- Tool calling
- Retry logic
"""
def __init__(
self,
api_key: str,
api_base: str = "https://api.minimaxi.com/v1",
model: str = "MiniMax-M2.5",
retry_config: Optional[RetryConfig] = None,
):
"""Initialize OpenAI client.
Args:
api_key: API key for authentication
api_base: Base URL for the API (default: MiniMax OpenAI endpoint)
model: Model name to use (default: MiniMax-M2.5)
retry_config: Optional retry configuration
"""
super().__init__(api_key, api_base, model, retry_config)
# Initialize OpenAI client
self.client = AsyncOpenAI(
api_key=api_key,
base_url=api_base,
)
async def _make_api_request(
self,
api_messages: list[dict[str, Any]],
tools: Optional[list[Any]] = None,
) -> Any:
"""Execute API request (core method that can be retried).
Args:
api_messages: List of messages in OpenAI format
tools: Optional list of tools
Returns:
OpenAI ChatCompletion response (full response including usage)
Raises:
Exception: API call failed
"""
params = {
"model": self.model,
"messages": api_messages,
# Enable reasoning_split to separate thinking content
"extra_body": {"reasoning_split": True},
}
if tools:
params["tools"] = self._convert_tools(tools)
# Use OpenAI SDK's chat.completions.create
response = await self.client.chat.completions.create(**params)
# Return full response to access usage info
return response
def _convert_tools(self, tools: list[Any]) -> list[dict[str, Any]]:
"""Convert tools to OpenAI format.
Args:
tools: List of Tool objects or dicts
Returns:
List of tools in OpenAI dict format
"""
result = []
for tool in tools:
if isinstance(tool, dict):
# If already a dict, check if it's in OpenAI format
if "type" in tool and tool["type"] == "function":
result.append(tool)
else:
# Assume it's in Anthropic format, convert to OpenAI
result.append(
{
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["input_schema"],
},
}
)
elif hasattr(tool, "to_openai_schema"):
# Tool object with to_openai_schema method
result.append(tool.to_openai_schema())
else:
raise TypeError(f"Unsupported tool type: {type(tool)}")
return result
def _convert_messages(self, messages: list[Message]) -> tuple[Optional[str], list[dict[str, Any]]]:
"""Convert internal messages to OpenAI format.
Args:
messages: List of internal Message objects
Returns:
Tuple of (system_message, api_messages)
Note: OpenAI includes system message in the messages array
"""
api_messages = []
for msg in messages:
if msg.role == "system":
# OpenAI includes system message in messages array
api_messages.append({"role": "system", "content": msg.content})
continue
# For user messages
if msg.role == "user":
api_messages.append({"role": "user", "content": msg.content})
# For assistant messages
elif msg.role == "assistant":
assistant_msg = {"role": "assistant"}
# Add content if present
if msg.content:
assistant_msg["content"] = msg.content
# Add tool calls if present
if msg.tool_calls:
tool_calls_list = []
for tool_call in msg.tool_calls:
tool_calls_list.append(
{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_call.function.name,
"arguments": json.dumps(tool_call.function.arguments),
},
}
)
assistant_msg["tool_calls"] = tool_calls_list
# IMPORTANT: Add reasoning_details if thinking is present
# This is CRITICAL for Interleaved Thinking to work properly!
# The complete response_message (including reasoning_details) must be
# preserved in Message History and passed back to the model in the next turn.
# This ensures the model's chain of thought is not interrupted.
if msg.thinking:
assistant_msg["reasoning_details"] = [{"text": msg.thinking}]
api_messages.append(assistant_msg)
# For tool result messages
elif msg.role == "tool":
api_messages.append(
{
"role": "tool",
"tool_call_id": msg.tool_call_id,
"content": msg.content,
}
)
return None, api_messages
def _prepare_request(
self,
messages: list[Message],
tools: Optional[list[Any]] = None,
) -> dict[str, Any]:
"""Prepare the request for OpenAI API.
Args:
messages: List of conversation messages
tools: Optional list of available tools
Returns:
Dictionary containing request parameters
"""
_, api_messages = self._convert_messages(messages)
return {
"api_messages": api_messages,
"tools": tools,
}
def _parse_response(self, response: Any) -> LLMResponse:
"""Parse OpenAI response into LLMResponse.
Args:
response: OpenAI ChatCompletion response (full response object)
Returns:
LLMResponse object
"""
# Get message from response
message = response.choices[0].message
# Extract text content
text_content = message.content or ""
# Extract thinking content from reasoning_details
thinking_content = ""
if hasattr(message, "reasoning_details") and message.reasoning_details:
# reasoning_details is a list of reasoning blocks
for detail in message.reasoning_details:
if hasattr(detail, "text"):
thinking_content += detail.text
# Extract tool calls
tool_calls = []
if message.tool_calls:
for tool_call in message.tool_calls:
# Parse arguments from JSON string
arguments = json.loads(tool_call.function.arguments)
tool_calls.append(
ToolCall(
id=tool_call.id,
type="function",
function=FunctionCall(
name=tool_call.function.name,
arguments=arguments,
),
)
)
# Extract token usage from response
usage = None
if hasattr(response, "usage") and response.usage:
usage = TokenUsage(
prompt_tokens=response.usage.prompt_tokens or 0,
completion_tokens=response.usage.completion_tokens or 0,
total_tokens=response.usage.total_tokens or 0,
)
return LLMResponse(
content=text_content,
thinking=thinking_content if thinking_content else None,
tool_calls=tool_calls if tool_calls else None,
finish_reason="stop", # OpenAI doesn't provide finish_reason in the message
usage=usage,
)
async def generate(
self,
messages: list[Message],
tools: Optional[list[Any]] = None,
) -> LLMResponse:
"""Generate response from OpenAI LLM.
Args:
messages: List of conversation messages
tools: Optional list of available tools
Returns:
LLMResponse containing the generated content
"""
# Prepare request
request_params = self._prepare_request(messages, tools)
# Make API request with retry logic
if self.retry_config.enabled:
# Apply retry logic
retry_decorator = async_retry(config=self.retry_config, on_retry=self.retry_callback)
api_call = retry_decorator(self._make_api_request)
response = await api_call(
request_params["api_messages"],
request_params["tools"],
)
else:
# Don't use retry
response = await self._make_api_request(
request_params["api_messages"],
request_params["tools"],
)
# Parse and return response
return self._parse_response(response)