| 1 | from __future__ import annotations |
| 2 | |
| 3 | from dataclasses import dataclass, field |
| 4 | import json |
| 5 | from typing import Any |
| 6 | |
| 7 | |
| 8 | RESPONSE_METADATA_KEY = "responses" |
| 9 | LOCAL_FUNCTION_TOOL_TYPES = {"function_call"} |
| 10 | TEXT_OUTPUT_TYPES = {"message"} |
| 11 | REASONING_OUTPUT_TYPES = {"reasoning"} |
| 12 | |
| 13 | |
| 14 | @dataclass |
| 15 | class ResponseItem: |
| 16 | type: str |
| 17 | data: dict[str, Any] = field(default_factory=dict) |
| 18 | |
| 19 | @classmethod |
| 20 | def from_any(cls, item: Any) -> "ResponseItem": |
| 21 | data = object_to_dict(item) |
| 22 | return cls(type=str(data.get("type") or ""), data=data) |
| 23 | |
| 24 | def to_dict(self) -> dict[str, Any]: |
| 25 | return dict(self.data) |
| 26 | |
| 27 | |
| 28 | @dataclass |
| 29 | class ResponseFunctionCall: |
| 30 | name: str |
| 31 | arguments: dict[str, Any] |
| 32 | call_id: str |
| 33 | item_id: str = "" |
| 34 | raw: dict[str, Any] = field(default_factory=dict) |
| 35 | |
| 36 | @classmethod |
| 37 | def from_item(cls, item: ResponseItem) -> "ResponseFunctionCall | None": |
| 38 | if item.type != "function_call": |
| 39 | return None |
| 40 | name = str(item.data.get("name") or "") |
| 41 | if not name: |
| 42 | return None |
| 43 | return cls( |
| 44 | name=name, |
| 45 | arguments=parse_arguments(item.data.get("arguments")), |
| 46 | call_id=str(item.data.get("call_id") or item.data.get("id") or ""), |
| 47 | item_id=str(item.data.get("id") or ""), |
| 48 | raw=dict(item.data), |
| 49 | ) |
| 50 | |
| 51 | |
| 52 | @dataclass |
| 53 | class LLMResult: |
| 54 | response: str = "" |
| 55 | reasoning: str = "" |
| 56 | response_id: str = "" |
| 57 | previous_response_id: str = "" |
| 58 | input_items: list[dict[str, Any]] = field(default_factory=list) |
| 59 | output_items: list[ResponseItem] = field(default_factory=list) |
| 60 | provider_model_key: str = "" |
| 61 | mode: str = "responses" |
| 62 | state: str = "provider" |
| 63 | usage: dict[str, Any] = field(default_factory=dict) |
| 64 | raw: dict[str, Any] = field(default_factory=dict) |
| 65 | capability: dict[str, Any] = field(default_factory=dict) |
| 66 | |
| 67 | @classmethod |
| 68 | def from_dict(cls, data: dict[str, Any] | None) -> "LLMResult": |
| 69 | data = data or {} |
| 70 | return cls( |
| 71 | response=str(data.get("response") or ""), |
| 72 | reasoning=str(data.get("reasoning") or ""), |
| 73 | response_id=str(data.get("response_id") or ""), |
| 74 | previous_response_id=str(data.get("previous_response_id") or ""), |
| 75 | input_items=list(data.get("input_items") or []), |
| 76 | output_items=[ |
| 77 | ResponseItem.from_any(item) for item in data.get("output_items") or [] |
| 78 | ], |
| 79 | provider_model_key=str(data.get("provider_model_key") or ""), |
| 80 | mode=str(data.get("mode") or "responses"), |
| 81 | state=str(data.get("state") or "provider"), |
| 82 | usage=object_to_dict(data.get("usage") or {}), |
| 83 | raw=object_to_dict(data.get("raw") or {}), |
| 84 | capability=object_to_dict(data.get("capability") or {}), |
| 85 | ) |
| 86 | |
| 87 | @classmethod |
| 88 | def from_response( |
| 89 | cls, |
| 90 | response: Any, |
| 91 | *, |
| 92 | input_items: list[dict[str, Any]] | None = None, |
| 93 | previous_response_id: str = "", |
| 94 | provider_model_key: str = "", |
| 95 | mode: str = "responses", |
| 96 | state: str = "provider", |
| 97 | capability: dict[str, Any] | None = None, |
| 98 | ) -> "LLMResult": |
| 99 | raw = object_to_dict(response) |
| 100 | output_items = [ResponseItem.from_any(item) for item in as_list(raw.get("output"))] |
| 101 | result = cls( |
| 102 | response_id=str(raw.get("id") or ""), |
| 103 | previous_response_id=str( |
| 104 | raw.get("previous_response_id") or previous_response_id or "" |
| 105 | ), |
| 106 | input_items=list(input_items or []), |
| 107 | output_items=output_items, |
| 108 | provider_model_key=provider_model_key, |
| 109 | mode=mode, |
| 110 | state=state, |
| 111 | usage=object_to_dict(raw.get("usage") or {}), |
| 112 | raw=raw, |
| 113 | capability=dict(capability or {}), |
| 114 | ) |
| 115 | result.response = output_text(raw, output_items) |
| 116 | result.reasoning = reasoning_text(output_items) |
| 117 | if not result.response and result.function_calls: |
| 118 | result.response = result.function_calls_text() |
| 119 | return result |
| 120 | |
| 121 | @classmethod |
| 122 | def from_chat( |
| 123 | cls, |
| 124 | *, |
| 125 | response: str, |
| 126 | reasoning: str = "", |
| 127 | usage: dict[str, Any] | None = None, |
| 128 | input_items: list[dict[str, Any]] | None = None, |
| 129 | output_items: list[dict[str, Any]] | None = None, |
| 130 | provider_model_key: str = "", |
| 131 | capability: dict[str, Any] | None = None, |
| 132 | ) -> "LLMResult": |
| 133 | items = [ResponseItem.from_any(item) for item in output_items or []] |
| 134 | if response and not items: |
| 135 | items.append( |
| 136 | ResponseItem( |
| 137 | type="message", |
| 138 | data={ |
| 139 | "type": "message", |
| 140 | "role": "assistant", |
| 141 | "content": [{"type": "output_text", "text": response}], |
| 142 | }, |
| 143 | ) |
| 144 | ) |
| 145 | if reasoning: |
| 146 | items.insert( |
| 147 | 0, |
| 148 | ResponseItem( |
| 149 | type="reasoning", |
| 150 | data={ |
| 151 | "type": "reasoning", |
| 152 | "summary": [{"type": "summary_text", "text": reasoning}], |
| 153 | }, |
| 154 | ), |
| 155 | ) |
| 156 | result = cls( |
| 157 | response=response, |
| 158 | reasoning=reasoning, |
| 159 | input_items=list(input_items or []), |
| 160 | output_items=items, |
| 161 | provider_model_key=provider_model_key, |
| 162 | mode="chat_completions", |
| 163 | state="off", |
| 164 | usage=object_to_dict(usage or {}), |
| 165 | capability=dict(capability or {}), |
| 166 | ) |
| 167 | if not result.response and result.function_calls: |
| 168 | result.response = result.function_calls_text() |
| 169 | return result |
| 170 | |
| 171 | @property |
| 172 | def function_calls(self) -> list[ResponseFunctionCall]: |
| 173 | calls: list[ResponseFunctionCall] = [] |
| 174 | for item in self.output_items: |
| 175 | call = ResponseFunctionCall.from_item(item) |
| 176 | if call: |
| 177 | calls.append(call) |
| 178 | return calls |
| 179 | |
| 180 | @property |
| 181 | def builtin_items(self) -> list[ResponseItem]: |
| 182 | return [ |
| 183 | item |
| 184 | for item in self.output_items |
| 185 | if item.type |
| 186 | and item.type not in TEXT_OUTPUT_TYPES |
| 187 | and item.type not in REASONING_OUTPUT_TYPES |
| 188 | and item.type not in LOCAL_FUNCTION_TOOL_TYPES |
| 189 | ] |
| 190 | |
| 191 | def function_calls_text(self) -> str: |
| 192 | calls = [ |
| 193 | {"tool_name": call.name, "tool_args": call.arguments} |
| 194 | for call in self.function_calls |
| 195 | ] |
| 196 | if not calls: |
| 197 | return "" |
| 198 | if len(calls) == 1: |
| 199 | return json.dumps(calls[0], ensure_ascii=False) |
| 200 | return json.dumps( |
| 201 | {"tool_name": "parallel_tool_calls", "tool_args": {"calls": calls}}, |
| 202 | ensure_ascii=False, |
| 203 | ) |
| 204 | |
| 205 | def to_dict(self) -> dict[str, Any]: |
| 206 | return { |
| 207 | "response": self.response, |
| 208 | "reasoning": self.reasoning, |
| 209 | "response_id": self.response_id, |
| 210 | "previous_response_id": self.previous_response_id, |
| 211 | "input_items": self.input_items, |
| 212 | "output_items": [item.to_dict() for item in self.output_items], |
| 213 | "provider_model_key": self.provider_model_key, |
| 214 | "mode": self.mode, |
| 215 | "state": self.state, |
| 216 | "usage": self.usage, |
| 217 | "raw": self.raw, |
| 218 | "capability": self.capability, |
| 219 | } |
| 220 | |
| 221 | def metadata(self) -> dict[str, Any]: |
| 222 | return { |
| 223 | RESPONSE_METADATA_KEY: { |
| 224 | "response_id": self.response_id, |
| 225 | "previous_response_id": self.previous_response_id, |
| 226 | "output_items": [item.to_dict() for item in self.output_items], |
| 227 | "provider_model_key": self.provider_model_key, |
| 228 | "mode": self.mode, |
| 229 | "state": self.state, |
| 230 | "usage": self.usage, |
| 231 | "capability": self.capability, |
| 232 | } |
| 233 | } |
| 234 | |
| 235 | |
| 236 | def function_call_output_item( |
| 237 | call_id: str, |
| 238 | output: str, |
| 239 | *, |
| 240 | acknowledged_safety_checks: list[dict[str, Any]] | None = None, |
| 241 | ) -> dict[str, Any]: |
| 242 | item: dict[str, Any] = { |
| 243 | "type": "function_call_output", |
| 244 | "call_id": str(call_id or ""), |
| 245 | "output": output, |
| 246 | } |
| 247 | if acknowledged_safety_checks: |
| 248 | item["acknowledged_safety_checks"] = acknowledged_safety_checks |
| 249 | return item |
| 250 | |
| 251 | |
| 252 | def metadata_from_llm_result(result: LLMResult | None) -> dict[str, Any]: |
| 253 | return result.metadata() if result else {} |
| 254 | |
| 255 | |
| 256 | def result_from_metadata(metadata: dict[str, Any] | None) -> LLMResult | None: |
| 257 | if not isinstance(metadata, dict): |
| 258 | return None |
| 259 | data = metadata.get(RESPONSE_METADATA_KEY) |
| 260 | if not isinstance(data, dict): |
| 261 | return None |
| 262 | return LLMResult.from_dict(data) |
| 263 | |
| 264 | |
| 265 | def object_to_dict(obj: Any) -> dict[str, Any]: |
| 266 | if isinstance(obj, dict): |
| 267 | return dict(obj) |
| 268 | if hasattr(obj, "model_dump"): |
| 269 | dumped = obj.model_dump() |
| 270 | return dict(dumped) if isinstance(dumped, dict) else {} |
| 271 | if hasattr(obj, "dict"): |
| 272 | dumped = obj.dict() |
| 273 | return dict(dumped) if isinstance(dumped, dict) else {} |
| 274 | return {} |
| 275 | |
| 276 | |
| 277 | def as_list(value: Any) -> list[Any]: |
| 278 | return value if isinstance(value, list) else [] |
| 279 | |
| 280 | |
| 281 | def output_text(raw: dict[str, Any], output_items: list[ResponseItem]) -> str: |
| 282 | direct = raw.get("output_text") |
| 283 | if isinstance(direct, str): |
| 284 | return direct |
| 285 | pieces: list[str] = [] |
| 286 | for item in output_items: |
| 287 | if item.type != "message": |
| 288 | continue |
| 289 | for block in as_list(item.data.get("content")): |
| 290 | if not isinstance(block, dict): |
| 291 | continue |
| 292 | block_type = block.get("type") |
| 293 | if block_type in {"output_text", "text", "input_text"}: |
| 294 | text = block.get("text") |
| 295 | if isinstance(text, str): |
| 296 | pieces.append(text) |
| 297 | elif block_type == "refusal": |
| 298 | refusal = block.get("refusal") |
| 299 | if isinstance(refusal, str): |
| 300 | pieces.append(refusal) |
| 301 | return "".join(pieces) |
| 302 | |
| 303 | |
| 304 | def reasoning_text(output_items: list[ResponseItem]) -> str: |
| 305 | pieces: list[str] = [] |
| 306 | for item in output_items: |
| 307 | if item.type != "reasoning": |
| 308 | continue |
| 309 | for block in as_list(item.data.get("summary")): |
| 310 | if isinstance(block, dict): |
| 311 | text = block.get("text") or block.get("reasoning") |
| 312 | if isinstance(text, str): |
| 313 | pieces.append(text) |
| 314 | elif isinstance(block, str): |
| 315 | pieces.append(block) |
| 316 | return "".join(pieces) |
| 317 | |
| 318 | |
| 319 | def parse_arguments(raw_arguments: Any) -> dict[str, Any]: |
| 320 | if isinstance(raw_arguments, dict): |
| 321 | return raw_arguments |
| 322 | if isinstance(raw_arguments, str): |
| 323 | try: |
| 324 | parsed = json.loads(raw_arguments or "{}") |
| 325 | except Exception: |
| 326 | parsed = {"arguments": raw_arguments} |
| 327 | else: |
| 328 | parsed = {"arguments": raw_arguments} |
| 329 | return parsed if isinstance(parsed, dict) else {"arguments": parsed} |