def convert__openai_responses_to_openai_chat__response(
response: openai_models.ResponseObject,
) -> openai_models.ChatCompletionResponse:
"""Convert an OpenAI ResponseObject to a ChatCompletionResponse."""
text_segments: list[str] = []
added_reasoning: set[tuple[str, str]] = set()
tool_calls: list[openai_models.ToolCall] = []
for item in response.output or []:
logger.debug(
"convert_responses_to_chat_response_item", item_type=_get_attr(item, "type")
)
item_type = _get_attr(item, "type")
if item_type == "reasoning":
for segment in _extract_reasoning_blocks(item):
signature = segment.signature
thinking_text = segment.thinking
logger.debug(
"convert_responses_to_chat_reasoning_block",
signature=signature,
text_snippet=(thinking_text[:30] + "...")
if thinking_text and len(thinking_text) > 30
else thinking_text,
)
if thinking_text:
key = (signature or "", thinking_text)
if key not in added_reasoning:
text_segments.append(_wrap_thinking(signature, thinking_text))
added_reasoning.add(key)
elif item_type == "message":
parts: list[str] = []
content_list = _get_attr(item, "content")
if isinstance(content_list, list):
for part in content_list:
part_type = _get_attr(part, "type")
if part_type == "output_text":
text_val = _get_attr(part, "text")
if isinstance(text_val, str):
parts.append(text_val)
elif isinstance(part, str):
parts.append(part)
elif isinstance(content_list, str):
parts.append(content_list)
if parts:
text_segments.append("".join(parts))
elif item_type == "function_call":
function_block = _get_attr(item, "function")
name = _get_attr(function_block, "name") or _get_attr(item, "name")
arguments_value: Any = _get_attr(item, "arguments")
if arguments_value is None and isinstance(function_block, dict):
arguments_value = function_block.get("arguments")
if not isinstance(name, str) or not name:
continue
if isinstance(arguments_value, dict):
arguments_str = json.dumps(arguments_value)
elif isinstance(arguments_value, str):
arguments_str = arguments_value
else:
arguments_str = json.dumps(arguments_value or {})
tool_calls.append(
openai_models.ToolCall(
id=_get_attr(item, "id")
or _get_attr(item, "call_id")
or f"call_{len(tool_calls)}",
type="function",
function=openai_models.FunctionCall(
name=name,
arguments=arguments_str,
),
)
)
text_content = "".join(text_segments)
usage = None
if response.usage:
usage = convert__openai_responses_usage_to_openai_completion__usage(
response.usage
)
finish_reason: Literal["stop", "length", "tool_calls", "content_filter"] = (
"tool_calls" if tool_calls else "stop"
)
return openai_models.ChatCompletionResponse(
id=response.id or "chatcmpl-resp",
choices=[
openai_models.Choice(
index=0,
message=openai_models.ResponseMessage(
role="assistant",
content=text_content,
tool_calls=tool_calls or None,
),
finish_reason=finish_reason,
)
],
created=0,
model=response.model or "",
object="chat.completion",
usage=usage
or openai_models.CompletionUsage(
prompt_tokens=0, completion_tokens=0, total_tokens=0
),
)