| 14 | |
| 15 | |
| 16 | class OpenAIChat: |
| 17 | |
| 18 | def __init__(self, |
| 19 | api_key: str = None, |
| 20 | base_url: str = None, |
| 21 | model: str = None, |
| 22 | **kwargs): |
| 23 | """ |
| 24 | Initialize the OpenAIChat client. |
| 25 | """ |
| 26 | self._client = OpenAI( |
| 27 | api_key=api_key, |
| 28 | base_url=base_url, |
| 29 | ) |
| 30 | |
| 31 | self._model = model |
| 32 | self._kwargs = kwargs |
| 33 | |
| 34 | def chat(self, |
| 35 | messages: List[Dict[str, Any]], |
| 36 | tools: List[Dict[str, Any]] = None, |
| 37 | **kwargs) -> Dict[str, Any]: |
| 38 | |
| 39 | completion: ChatCompletion = self._client.chat.completions.create( |
| 40 | messages=messages, model=self._model, tools=tools, **kwargs) |
| 41 | |
| 42 | res_d: Dict[str, Any] = dict( |
| 43 | role='assistant', |
| 44 | reasoning_content='', |
| 45 | content=completion.choices[0].message.content, |
| 46 | tool_calls=completion.choices[0].message.tool_calls if hasattr( |
| 47 | completion.choices[0].message, 'tool_calls') else [], |
| 48 | finish_reason=completion.choices[0]. |
| 49 | finish_reason, # 'stop', 'tool_calls', 'length', None |
| 50 | usage=completion.usage.to_dict(), |
| 51 | ) |
| 52 | |
| 53 | return res_d |
| 54 | |
| 55 | def chat_stream(self, |
| 56 | messages: List[Dict[str, Any]], |
| 57 | tools: List[Dict[str, Any]] = None, |
| 58 | **kwargs): |
| 59 | """ |
| 60 | Get chat response from OpenAI API using streaming. |
| 61 | |
| 62 | messages: |
| 63 | A list of dictionaries representing the chat messages. |
| 64 | Each dictionary should have 'role' (e.g., 'user', 'assistant') and 'content'. |
| 65 | Fully compatible with OpenAI's chat completion API. |
| 66 | [ |
| 67 | { |
| 68 | "role": str, # Required, one of 'user', 'assistant', 'system', 'tool' |
| 69 | "content": Optional[str], # Optional, required if role is 'user' or 'assistant' |
| 70 | "name": Optional[str], # Optional, required if role is 'tool' |
| 71 | "tool_calls": Optional[List[Dict]], # Optional, required if role is 'tool' |
| 72 | "tool_call_id": Optional[str], # Optional, required if role is 'tool' |
| 73 | "function_call": Optional[Dict], # Deprecated, use 'tool_calls' instead |
no outgoing calls
no test coverage detected