| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Streaming SSE event builders for the Responses API. |
| 4 | |
| 5 | Pure functions that translate streaming state + delta data into |
| 6 | OpenAI Response API SSE events. Used by the streaming event |
| 7 | processors in serving.py. |
| 8 | |
| 9 | The file is organized as: |
| 10 | 1. StreamingState dataclass + utility helpers |
| 11 | 2. Shared leaf helpers — delta events (take plain strings, no context) |
| 12 | 3. Shared leaf helpers — done events (take plain strings, no context) |
| 13 | 4. Harmony-specific dispatchers (route ctx/previous_item → leaf helpers) |
| 14 | 5. Harmony-specific tool lifecycle helpers |
| 15 | """ |
| 16 | |
| 17 | import json |
| 18 | from collections.abc import Callable |
| 19 | from dataclasses import dataclass, field |
| 20 | from enum import Enum, auto |
| 21 | from typing import Any, ClassVar, Final, NamedTuple |
| 22 | |
| 23 | from openai.types.responses import ( |
| 24 | ResponseCodeInterpreterCallCodeDeltaEvent, |
| 25 | ResponseCodeInterpreterCallCodeDoneEvent, |
| 26 | ResponseCodeInterpreterCallCompletedEvent, |
| 27 | ResponseCodeInterpreterCallInProgressEvent, |
| 28 | ResponseCodeInterpreterCallInterpretingEvent, |
| 29 | ResponseCodeInterpreterToolCallParam, |
| 30 | ResponseContentPartAddedEvent, |
| 31 | ResponseContentPartDoneEvent, |
| 32 | ResponseFunctionCallArgumentsDeltaEvent, |
| 33 | ResponseFunctionCallArgumentsDoneEvent, |
| 34 | ResponseFunctionToolCall, |
| 35 | ResponseFunctionToolCallItem, |
| 36 | ResponseFunctionWebSearch, |
| 37 | ResponseMcpCallArgumentsDeltaEvent, |
| 38 | ResponseMcpCallArgumentsDoneEvent, |
| 39 | ResponseMcpCallCompletedEvent, |
| 40 | ResponseMcpCallInProgressEvent, |
| 41 | ResponseOutputItem, |
| 42 | ResponseOutputItemAddedEvent, |
| 43 | ResponseOutputItemDoneEvent, |
| 44 | ResponseOutputMessage, |
| 45 | ResponseOutputText, |
| 46 | ResponseReasoningItem, |
| 47 | ResponseReasoningTextDeltaEvent, |
| 48 | ResponseReasoningTextDoneEvent, |
| 49 | ResponseTextDeltaEvent, |
| 50 | ResponseTextDoneEvent, |
| 51 | ResponseWebSearchCallCompletedEvent, |
| 52 | ResponseWebSearchCallInProgressEvent, |
| 53 | ResponseWebSearchCallSearchingEvent, |
| 54 | response_function_web_search, |
| 55 | response_output_text, |
| 56 | response_text_delta_event, |
| 57 | ) |
| 58 | from openai.types.responses.response_output_item import McpCall |
| 59 | from openai.types.responses.response_reasoning_item import ( |
| 60 | Content as ResponseReasoningTextContent, |
| 61 | ) |
| 62 | from openai.types.responses.tool import Tool |
| 63 | from openai_harmony import Message as HarmonyMessage |
| 64 | |
| 65 | from vllm.entrypoints.chat_utils import make_tool_call_id |
| 66 | from vllm.entrypoints.generate.base.protocol import DeltaMessage, DeltaToolCall |
| 67 | from vllm.entrypoints.mcp.tool_server import ToolServer |
| 68 | from vllm.entrypoints.openai.parser.harmony_utils import ( |
| 69 | extract_function_from_recipient, |
| 70 | is_function_recipient, |
| 71 | ) |
| 72 | from vllm.entrypoints.openai.responses.protocol import ( |
| 73 | StreamingResponsesResponse, |
| 74 | ) |
| 75 | from vllm.entrypoints.openai.responses.utils import ( |
| 76 | build_responses_tool_call_name_map, |
| 77 | resolve_responses_tool_call_name, |
| 78 | ) |
| 79 | from vllm.outputs import CompletionOutput |
| 80 | from vllm.parser.harmony import Segment |
| 81 | from vllm.utils import random_uuid |
| 82 | |
| 83 | TOOL_NAME_TO_MCP_SERVER_LABEL: Final[dict[str, str]] = { |
| 84 | "python": "code_interpreter", |
| 85 | "container": "container", |
| 86 | "browser": "web_search_preview", |
| 87 | } |
| 88 | |
| 89 | |
| 90 | def _resolve_mcp_name_label(recipient: str) -> tuple[str, str]: |
| 91 | """Resolve MCP tool name and server label from a recipient string. |
| 92 | |
| 93 | - ``mcp.*`` recipients: strip prefix, use the bare name as both |
| 94 | name and server_label. |
| 95 | - Everything else: use the recipient as the name and look up the |
| 96 | server_label in TOOL_NAME_TO_MCP_SERVER_LABEL. |
| 97 | """ |
| 98 | if recipient.startswith("mcp."): |
| 99 | name = recipient[len("mcp.") :] |
| 100 | return name, name |
| 101 | return recipient, TOOL_NAME_TO_MCP_SERVER_LABEL.get(recipient, recipient) |
| 102 | |
| 103 | |
| 104 | @dataclass |
| 105 | class StreamingState: |
| 106 | """Mutable state for streaming event processing.""" |
| 107 | |
| 108 | current_content_index: int = -1 |
| 109 | current_output_index: int = 0 |
| 110 | current_item_id: str = "" |
| 111 | current_call_id: str = "" |
| 112 | sent_output_item_added: bool = False |
| 113 | is_first_function_call_delta: bool = False |
| 114 | |
| 115 | def reset_for_new_item(self) -> None: |
| 116 | """Reset state when expecting a new output item.""" |
| 117 | self.current_output_index += 1 |
| 118 | self.current_content_index = -1 |
| 119 | self.sent_output_item_added = False |
| 120 | self.is_first_function_call_delta = False |
| 121 | self.current_call_id = "" |
| 122 | |
| 123 | |
| 124 | def is_mcp_tool_by_namespace( |
| 125 | recipient: str | None, |
| 126 | allowed_function_tool_names: frozenset[str] | None = None, |
| 127 | ) -> bool: |
| 128 | """Determine if a tool call is an MCP tool based on recipient prefix. |
| 129 | |
| 130 | Inverse of :func:`is_function_recipient` — everything that is not |
| 131 | a function call is an MCP tool. |
| 132 | """ |
| 133 | if recipient is None: |
| 134 | return False |
| 135 | return not is_function_recipient(recipient, allowed_function_tool_names) |
| 136 | |
| 137 | |
| 138 | # ===================================================================== |
| 139 | # Shared leaf helpers — delta events |
| 140 | # ===================================================================== |
| 141 | |
| 142 | |
| 143 | def emit_text_delta_events( |
| 144 | delta: str, |
| 145 | state: StreamingState, |
| 146 | ) -> list[StreamingResponsesResponse]: |
| 147 | """Emit events for text content delta streaming.""" |
| 148 | events: list[StreamingResponsesResponse] = [] |
| 149 | if not state.sent_output_item_added: |
| 150 | state.sent_output_item_added = True |
| 151 | state.current_item_id = f"msg_{random_uuid()}" |
| 152 | events.append( |
| 153 | ResponseOutputItemAddedEvent( |
| 154 | type="response.output_item.added", |
| 155 | sequence_number=-1, |
| 156 | output_index=state.current_output_index, |
| 157 | item=ResponseOutputMessage( |
| 158 | id=state.current_item_id, |
| 159 | type="message", |
| 160 | role="assistant", |
| 161 | content=[], |
| 162 | status="in_progress", |
| 163 | ), |
| 164 | ) |
| 165 | ) |
| 166 | state.current_content_index += 1 |
| 167 | events.append( |
| 168 | ResponseContentPartAddedEvent( |
| 169 | type="response.content_part.added", |
| 170 | sequence_number=-1, |
| 171 | output_index=state.current_output_index, |
| 172 | item_id=state.current_item_id, |
| 173 | content_index=state.current_content_index, |
| 174 | part=ResponseOutputText( |
| 175 | type="output_text", |
| 176 | text="", |
| 177 | annotations=[], |
| 178 | logprobs=[], |
| 179 | ), |
| 180 | ) |
| 181 | ) |
| 182 | events.append( |
| 183 | ResponseTextDeltaEvent( |
| 184 | type="response.output_text.delta", |
| 185 | sequence_number=-1, |
| 186 | content_index=state.current_content_index, |
| 187 | output_index=state.current_output_index, |
| 188 | item_id=state.current_item_id, |
| 189 | delta=delta, |
| 190 | # TODO, use logprobs from ctx.last_request_output |
| 191 | logprobs=[], |
| 192 | ) |
| 193 | ) |
| 194 | return events |
| 195 | |
| 196 | |
| 197 | def emit_reasoning_delta_events( |
| 198 | delta: str, |
| 199 | state: StreamingState, |
| 200 | ) -> list[StreamingResponsesResponse]: |
| 201 | """Emit events for reasoning text delta streaming.""" |
| 202 | events: list[StreamingResponsesResponse] = [] |
| 203 | if not state.sent_output_item_added: |
| 204 | state.sent_output_item_added = True |
| 205 | state.current_item_id = f"msg_{random_uuid()}" |
| 206 | events.append( |
| 207 | ResponseOutputItemAddedEvent( |
| 208 | type="response.output_item.added", |
| 209 | sequence_number=-1, |
| 210 | output_index=state.current_output_index, |
| 211 | item=ResponseReasoningItem( |
| 212 | type="reasoning", |
| 213 | id=state.current_item_id, |
| 214 | summary=[], |
| 215 | status="in_progress", |
| 216 | ), |
| 217 | ) |
| 218 | ) |
| 219 | state.current_content_index += 1 |
| 220 | events.append( |
| 221 | ResponseContentPartAddedEvent( |
| 222 | type="response.content_part.added", |
| 223 | sequence_number=-1, |
| 224 | output_index=state.current_output_index, |
| 225 | item_id=state.current_item_id, |
| 226 | content_index=state.current_content_index, |
| 227 | part={"text": "", "type": "reasoning_text"}, |
| 228 | ) |
| 229 | ) |
| 230 | events.append( |
| 231 | ResponseReasoningTextDeltaEvent( |
| 232 | type="response.reasoning_text.delta", |
| 233 | item_id=state.current_item_id, |
| 234 | output_index=state.current_output_index, |
| 235 | content_index=state.current_content_index, |
| 236 | delta=delta, |
| 237 | sequence_number=-1, |
| 238 | ) |
| 239 | ) |
| 240 | return events |
| 241 | |
| 242 | |
| 243 | def emit_function_call_delta_events( |
| 244 | delta: str, |
| 245 | function_name: str, |
| 246 | state: StreamingState, |
| 247 | ) -> list[StreamingResponsesResponse]: |
| 248 | """Emit events for function call argument deltas.""" |
| 249 | events: list[StreamingResponsesResponse] = [] |
| 250 | if state.is_first_function_call_delta is False: |
| 251 | state.is_first_function_call_delta = True |
| 252 | state.current_item_id = f"fc_{random_uuid()}" |
| 253 | state.current_call_id = f"call_{random_uuid()}" |
| 254 | tool_call_item = ResponseFunctionToolCall( |
| 255 | name=function_name, |
| 256 | type="function_call", |
| 257 | id=state.current_item_id, |
| 258 | call_id=state.current_call_id, |
| 259 | arguments="", |
| 260 | status="in_progress", |
| 261 | ) |
| 262 | events.append( |
| 263 | ResponseOutputItemAddedEvent( |
| 264 | type="response.output_item.added", |
| 265 | sequence_number=-1, |
| 266 | output_index=state.current_output_index, |
| 267 | item=tool_call_item, |
| 268 | ) |
| 269 | ) |
| 270 | # Always emit the delta (including on first call) |
| 271 | events.append( |
| 272 | ResponseFunctionCallArgumentsDeltaEvent( |
| 273 | item_id=state.current_item_id, |
| 274 | delta=delta, |
| 275 | output_index=state.current_output_index, |
| 276 | sequence_number=-1, |
| 277 | type="response.function_call_arguments.delta", |
| 278 | ) |
| 279 | ) |
| 280 | return events |
| 281 | |
| 282 | |
| 283 | def emit_mcp_delta_events( |
| 284 | delta: str, |
| 285 | state: StreamingState, |
| 286 | recipient: str, |
| 287 | ) -> list[StreamingResponsesResponse]: |
| 288 | """Emit events for MCP tool delta streaming.""" |
| 289 | name, server_label = _resolve_mcp_name_label(recipient) |
| 290 | events: list[StreamingResponsesResponse] = [] |
| 291 | if not state.sent_output_item_added: |
| 292 | state.sent_output_item_added = True |
| 293 | state.current_item_id = f"mcp_{random_uuid()}" |
| 294 | events.append( |
| 295 | ResponseOutputItemAddedEvent( |
| 296 | type="response.output_item.added", |
| 297 | sequence_number=-1, |
| 298 | output_index=state.current_output_index, |
| 299 | item=McpCall( |
| 300 | type="mcp_call", |
| 301 | id=state.current_item_id, |
| 302 | name=name, |
| 303 | arguments="", |
| 304 | server_label=server_label, |
| 305 | status="in_progress", |
| 306 | ), |
| 307 | ) |
| 308 | ) |
| 309 | events.append( |
| 310 | ResponseMcpCallInProgressEvent( |
| 311 | type="response.mcp_call.in_progress", |
| 312 | sequence_number=-1, |
| 313 | output_index=state.current_output_index, |
| 314 | item_id=state.current_item_id, |
| 315 | ) |
| 316 | ) |
| 317 | events.append( |
| 318 | ResponseMcpCallArgumentsDeltaEvent( |
| 319 | type="response.mcp_call_arguments.delta", |
| 320 | sequence_number=-1, |
| 321 | output_index=state.current_output_index, |
| 322 | item_id=state.current_item_id, |
| 323 | delta=delta, |
| 324 | ) |
| 325 | ) |
| 326 | return events |
| 327 | |
| 328 | |
| 329 | def emit_code_interpreter_delta_events( |
| 330 | delta: str, |
| 331 | state: StreamingState, |
| 332 | ) -> list[StreamingResponsesResponse]: |
| 333 | """Emit events for code interpreter delta streaming.""" |
| 334 | events: list[StreamingResponsesResponse] = [] |
| 335 | if not state.sent_output_item_added: |
| 336 | state.sent_output_item_added = True |
| 337 | state.current_item_id = f"tool_{random_uuid()}" |
| 338 | events.append( |
| 339 | ResponseOutputItemAddedEvent( |
| 340 | type="response.output_item.added", |
| 341 | sequence_number=-1, |
| 342 | output_index=state.current_output_index, |
| 343 | item=ResponseCodeInterpreterToolCallParam( |
| 344 | type="code_interpreter_call", |
| 345 | id=state.current_item_id, |
| 346 | code=None, |
| 347 | container_id="auto", |
| 348 | outputs=None, |
| 349 | status="in_progress", |
| 350 | ), |
| 351 | ) |
| 352 | ) |
| 353 | events.append( |
| 354 | ResponseCodeInterpreterCallInProgressEvent( |
| 355 | type="response.code_interpreter_call.in_progress", |
| 356 | sequence_number=-1, |
| 357 | output_index=state.current_output_index, |
| 358 | item_id=state.current_item_id, |
| 359 | ) |
| 360 | ) |
| 361 | events.append( |
| 362 | ResponseCodeInterpreterCallCodeDeltaEvent( |
| 363 | type="response.code_interpreter_call_code.delta", |
| 364 | sequence_number=-1, |
| 365 | output_index=state.current_output_index, |
| 366 | item_id=state.current_item_id, |
| 367 | delta=delta, |
| 368 | ) |
| 369 | ) |
| 370 | return events |
| 371 | |
| 372 | |
| 373 | # ===================================================================== |
| 374 | # Shared leaf helpers — done events |
| 375 | # ===================================================================== |
| 376 | |
| 377 | |
| 378 | def emit_text_output_done_events( |
| 379 | text: str, |
| 380 | state: StreamingState, |
| 381 | ) -> list[StreamingResponsesResponse]: |
| 382 | """Emit events when a final text output item completes.""" |
| 383 | text_content = ResponseOutputText( |
| 384 | type="output_text", |
| 385 | text=text, |
| 386 | annotations=[], |
| 387 | ) |
| 388 | events: list[StreamingResponsesResponse] = [] |
| 389 | events.append( |
| 390 | ResponseTextDoneEvent( |
| 391 | type="response.output_text.done", |
| 392 | sequence_number=-1, |
| 393 | output_index=state.current_output_index, |
| 394 | content_index=state.current_content_index, |
| 395 | text=text, |
| 396 | logprobs=[], |
| 397 | item_id=state.current_item_id, |
| 398 | ) |
| 399 | ) |
| 400 | events.append( |
| 401 | ResponseContentPartDoneEvent( |
| 402 | type="response.content_part.done", |
| 403 | sequence_number=-1, |
| 404 | item_id=state.current_item_id, |
| 405 | output_index=state.current_output_index, |
| 406 | content_index=state.current_content_index, |
| 407 | part=text_content, |
| 408 | ) |
| 409 | ) |
| 410 | events.append( |
| 411 | ResponseOutputItemDoneEvent( |
| 412 | type="response.output_item.done", |
| 413 | sequence_number=-1, |
| 414 | output_index=state.current_output_index, |
| 415 | item=ResponseOutputMessage( |
| 416 | id=state.current_item_id, |
| 417 | type="message", |
| 418 | role="assistant", |
| 419 | content=[text_content], |
| 420 | status="completed", |
| 421 | ), |
| 422 | ) |
| 423 | ) |
| 424 | return events |
| 425 | |
| 426 | |
| 427 | def emit_reasoning_done_events( |
| 428 | text: str, |
| 429 | state: StreamingState, |
| 430 | ) -> list[StreamingResponsesResponse]: |
| 431 | """Emit events when a reasoning (analysis) item completes.""" |
| 432 | content = ResponseReasoningTextContent( |
| 433 | text=text, |
| 434 | type="reasoning_text", |
| 435 | ) |
| 436 | reasoning_item = ResponseReasoningItem( |
| 437 | type="reasoning", |
| 438 | content=[content], |
| 439 | status="completed", |
| 440 | id=state.current_item_id, |
| 441 | summary=[], |
| 442 | ) |
| 443 | events: list[StreamingResponsesResponse] = [] |
| 444 | events.append( |
| 445 | ResponseReasoningTextDoneEvent( |
| 446 | type="response.reasoning_text.done", |
| 447 | item_id=state.current_item_id, |
| 448 | sequence_number=-1, |
| 449 | output_index=state.current_output_index, |
| 450 | content_index=state.current_content_index, |
| 451 | text=text, |
| 452 | ) |
| 453 | ) |
| 454 | events.append( |
| 455 | ResponseContentPartDoneEvent( |
| 456 | type="response.content_part.done", |
| 457 | sequence_number=-1, |
| 458 | item_id=state.current_item_id, |
| 459 | output_index=state.current_output_index, |
| 460 | content_index=state.current_content_index, |
| 461 | part=content.model_dump(), |
| 462 | ) |
| 463 | ) |
| 464 | events.append( |
| 465 | ResponseOutputItemDoneEvent( |
| 466 | type="response.output_item.done", |
| 467 | sequence_number=-1, |
| 468 | output_index=state.current_output_index, |
| 469 | item=reasoning_item, |
| 470 | ) |
| 471 | ) |
| 472 | return events |
| 473 | |
| 474 | |
| 475 | def emit_function_call_done_events( |
| 476 | function_name: str, |
| 477 | arguments: str, |
| 478 | state: StreamingState, |
| 479 | ) -> list[StreamingResponsesResponse]: |
| 480 | """Emit events when a function call completes.""" |
| 481 | events: list[StreamingResponsesResponse] = [] |
| 482 | events.append( |
| 483 | ResponseFunctionCallArgumentsDoneEvent( |
| 484 | type="response.function_call_arguments.done", |
| 485 | arguments=arguments, |
| 486 | name=function_name, |
| 487 | item_id=state.current_item_id, |
| 488 | output_index=state.current_output_index, |
| 489 | sequence_number=-1, |
| 490 | ) |
| 491 | ) |
| 492 | function_call_item = ResponseFunctionToolCall( |
| 493 | type="function_call", |
| 494 | arguments=arguments, |
| 495 | name=function_name, |
| 496 | id=state.current_item_id, |
| 497 | output_index=state.current_output_index, |
| 498 | sequence_number=-1, |
| 499 | call_id=state.current_call_id, |
| 500 | status="completed", |
| 501 | ) |
| 502 | events.append( |
| 503 | ResponseOutputItemDoneEvent( |
| 504 | type="response.output_item.done", |
| 505 | sequence_number=-1, |
| 506 | output_index=state.current_output_index, |
| 507 | item=function_call_item, |
| 508 | ) |
| 509 | ) |
| 510 | return events |
| 511 | |
| 512 | |
| 513 | def emit_mcp_completion_events( |
| 514 | recipient: str, |
| 515 | arguments: str, |
| 516 | state: StreamingState, |
| 517 | ) -> list[StreamingResponsesResponse]: |
| 518 | """Emit events when an MCP tool call completes.""" |
| 519 | name, server_label = _resolve_mcp_name_label(recipient) |
| 520 | events: list[StreamingResponsesResponse] = [] |
| 521 | events.append( |
| 522 | ResponseMcpCallArgumentsDoneEvent( |
| 523 | type="response.mcp_call_arguments.done", |
| 524 | arguments=arguments, |
| 525 | name=name, |
| 526 | item_id=state.current_item_id, |
| 527 | output_index=state.current_output_index, |
| 528 | sequence_number=-1, |
| 529 | ) |
| 530 | ) |
| 531 | events.append( |
| 532 | ResponseMcpCallCompletedEvent( |
| 533 | type="response.mcp_call.completed", |
| 534 | sequence_number=-1, |
| 535 | output_index=state.current_output_index, |
| 536 | item_id=state.current_item_id, |
| 537 | ) |
| 538 | ) |
| 539 | events.append( |
| 540 | ResponseOutputItemDoneEvent( |
| 541 | type="response.output_item.done", |
| 542 | sequence_number=-1, |
| 543 | output_index=state.current_output_index, |
| 544 | item=McpCall( |
| 545 | type="mcp_call", |
| 546 | arguments=arguments, |
| 547 | name=name, |
| 548 | id=state.current_item_id, |
| 549 | server_label=server_label, |
| 550 | status="completed", |
| 551 | ), |
| 552 | ) |
| 553 | ) |
| 554 | return events |
| 555 | |
| 556 | |
| 557 | # ===================================================================== |
| 558 | # Harmony-specific dispatchers |
| 559 | # ===================================================================== |
| 560 | |
| 561 | |
| 562 | def emit_content_delta_events( |
| 563 | segment: Segment, |
| 564 | state: StreamingState, |
| 565 | function_tool_names: frozenset[str] | None = None, |
| 566 | ) -> list[StreamingResponsesResponse]: |
| 567 | """Emit events for content delta streaming based on channel type. |
| 568 | |
| 569 | This is a Harmony-specific dispatcher that extracts values from the |
| 570 | latest append segment and delegates to shared leaf helpers. |
| 571 | """ |
| 572 | delta = segment.delta |
| 573 | if not delta: |
| 574 | return [] |
| 575 | |
| 576 | channel = segment.channel |
| 577 | recipient = segment.recipient |
| 578 | |
| 579 | if channel in ("final", "commentary") and recipient is None: |
| 580 | # Preambles (commentary with no recipient) and final messages |
| 581 | # are both user-visible text. |
| 582 | return emit_text_delta_events(delta, state) |
| 583 | elif channel == "analysis" and recipient is None: |
| 584 | return emit_reasoning_delta_events(delta, state) |
| 585 | elif recipient is not None: |
| 586 | fn_names = function_tool_names |
| 587 | if is_function_recipient(recipient, fn_names): |
| 588 | function_name = extract_function_from_recipient(recipient) |
| 589 | return emit_function_call_delta_events(delta, function_name, state) |
| 590 | elif recipient == "python": |
| 591 | return emit_code_interpreter_delta_events(delta, state) |
| 592 | elif recipient.startswith("mcp.") or is_mcp_tool_by_namespace( |
| 593 | recipient, fn_names |
| 594 | ): |
| 595 | return emit_mcp_delta_events(delta, state, recipient) |
| 596 | |
| 597 | return [] |
| 598 | |
| 599 | |
| 600 | def emit_previous_item_done_events( |
| 601 | previous_item: HarmonyMessage, |
| 602 | state: StreamingState, |
| 603 | function_tool_names: frozenset[str] | None = None, |
| 604 | ) -> list[StreamingResponsesResponse]: |
| 605 | """Emit done events for the previous item when expecting a new start. |
| 606 | |
| 607 | This is a Harmony-specific dispatcher that extracts values from the |
| 608 | Harmony parser's message object and delegates to shared leaf helpers. |
| 609 | """ |
| 610 | if not state.sent_output_item_added and not state.is_first_function_call_delta: |
| 611 | # Suppress done events for items had no delta and thus had no |
| 612 | # added/in-progress lifecycle events. This is a bug. |
| 613 | # TODO: Ensure added/in-progress events are emitted for zero-delta items. |
| 614 | return [] |
| 615 | |
| 616 | text = previous_item.content[0].text |
| 617 | if previous_item.recipient is not None: |
| 618 | # Deal with tool call |
| 619 | if is_function_recipient(previous_item.recipient, function_tool_names): |
| 620 | function_name = extract_function_from_recipient(previous_item.recipient) |
| 621 | return emit_function_call_done_events(function_name, text, state) |
| 622 | elif previous_item.recipient == "python": |
| 623 | return emit_code_interpreter_completion_events(previous_item, state) |
| 624 | elif ( |
| 625 | is_mcp_tool_by_namespace(previous_item.recipient, function_tool_names) |
| 626 | and state.current_item_id is not None |
| 627 | and state.current_item_id.startswith("mcp_") |
| 628 | ): |
| 629 | return emit_mcp_completion_events(previous_item.recipient, text, state) |
| 630 | elif previous_item.channel == "analysis": |
| 631 | return emit_reasoning_done_events(text, state) |
| 632 | elif previous_item.channel in ("commentary", "final"): |
| 633 | # Preambles (commentary with no recipient) and final messages |
| 634 | # are both user-visible text. |
| 635 | return emit_text_output_done_events(text, state) |
| 636 | return [] |
| 637 | |
| 638 | |
| 639 | # ===================================================================== |
| 640 | # Harmony-specific tool lifecycle helpers |
| 641 | # ===================================================================== |
| 642 | |
| 643 | |
| 644 | def emit_browser_tool_events( |
| 645 | previous_item: HarmonyMessage, |
| 646 | state: StreamingState, |
| 647 | ) -> list[StreamingResponsesResponse]: |
| 648 | """Emit events for browser tool calls (web search).""" |
| 649 | function_name = previous_item.recipient[len("browser.") :] |
| 650 | parsed_args = json.loads(previous_item.content[0].text) |
| 651 | action = None |
| 652 | |
| 653 | if function_name == "search": |
| 654 | action = response_function_web_search.ActionSearch( |
| 655 | type="search", |
| 656 | query=parsed_args["query"], |
| 657 | ) |
| 658 | elif function_name == "open": |
| 659 | action = response_function_web_search.ActionOpenPage( |
| 660 | type="open_page", |
| 661 | # TODO: translate to url |
| 662 | url=f"cursor:{parsed_args.get('cursor', '')}", |
| 663 | ) |
| 664 | elif function_name == "find": |
| 665 | action = response_function_web_search.ActionFind( |
| 666 | type="find_in_page", |
| 667 | pattern=parsed_args["pattern"], |
| 668 | # TODO: translate to url |
| 669 | url=f"cursor:{parsed_args.get('cursor', '')}", |
| 670 | ) |
| 671 | else: |
| 672 | raise ValueError(f"Unknown function name: {function_name}") |
| 673 | |
| 674 | state.current_item_id = f"tool_{random_uuid()}" |
| 675 | events: list[StreamingResponsesResponse] = [] |
| 676 | events.append( |
| 677 | ResponseOutputItemAddedEvent( |
| 678 | type="response.output_item.added", |
| 679 | sequence_number=-1, |
| 680 | output_index=state.current_output_index, |
| 681 | item=response_function_web_search.ResponseFunctionWebSearch( |
| 682 | # TODO: generate a unique id for web search call |
| 683 | type="web_search_call", |
| 684 | id=state.current_item_id, |
| 685 | action=action, |
| 686 | status="in_progress", |
| 687 | ), |
| 688 | ) |
| 689 | ) |
| 690 | events.append( |
| 691 | ResponseWebSearchCallInProgressEvent( |
| 692 | type="response.web_search_call.in_progress", |
| 693 | sequence_number=-1, |
| 694 | output_index=state.current_output_index, |
| 695 | item_id=state.current_item_id, |
| 696 | ) |
| 697 | ) |
| 698 | events.append( |
| 699 | ResponseWebSearchCallSearchingEvent( |
| 700 | type="response.web_search_call.searching", |
| 701 | sequence_number=-1, |
| 702 | output_index=state.current_output_index, |
| 703 | item_id=state.current_item_id, |
| 704 | ) |
| 705 | ) |
| 706 | # enqueue |
| 707 | events.append( |
| 708 | ResponseWebSearchCallCompletedEvent( |
| 709 | type="response.web_search_call.completed", |
| 710 | sequence_number=-1, |
| 711 | output_index=state.current_output_index, |
| 712 | item_id=state.current_item_id, |
| 713 | ) |
| 714 | ) |
| 715 | events.append( |
| 716 | ResponseOutputItemDoneEvent( |
| 717 | type="response.output_item.done", |
| 718 | sequence_number=-1, |
| 719 | output_index=state.current_output_index, |
| 720 | item=ResponseFunctionWebSearch( |
| 721 | type="web_search_call", |
| 722 | id=state.current_item_id, |
| 723 | action=action, |
| 724 | status="completed", |
| 725 | ), |
| 726 | ) |
| 727 | ) |
| 728 | return events |
| 729 | |
| 730 | |
| 731 | def emit_code_interpreter_completion_events( |
| 732 | previous_item: HarmonyMessage, |
| 733 | state: StreamingState, |
| 734 | ) -> list[StreamingResponsesResponse]: |
| 735 | """Emit events when code interpreter completes.""" |
| 736 | events: list[StreamingResponsesResponse] = [] |
| 737 | events.append( |
| 738 | ResponseCodeInterpreterCallCodeDoneEvent( |
| 739 | type="response.code_interpreter_call_code.done", |
| 740 | sequence_number=-1, |
| 741 | output_index=state.current_output_index, |
| 742 | item_id=state.current_item_id, |
| 743 | code=previous_item.content[0].text, |
| 744 | ) |
| 745 | ) |
| 746 | events.append( |
| 747 | ResponseCodeInterpreterCallInterpretingEvent( |
| 748 | type="response.code_interpreter_call.interpreting", |
| 749 | sequence_number=-1, |
| 750 | output_index=state.current_output_index, |
| 751 | item_id=state.current_item_id, |
| 752 | ) |
| 753 | ) |
| 754 | events.append( |
| 755 | ResponseCodeInterpreterCallCompletedEvent( |
| 756 | type="response.code_interpreter_call.completed", |
| 757 | sequence_number=-1, |
| 758 | output_index=state.current_output_index, |
| 759 | item_id=state.current_item_id, |
| 760 | ) |
| 761 | ) |
| 762 | events.append( |
| 763 | ResponseOutputItemDoneEvent( |
| 764 | type="response.output_item.done", |
| 765 | sequence_number=-1, |
| 766 | output_index=state.current_output_index, |
| 767 | item=ResponseCodeInterpreterToolCallParam( |
| 768 | type="code_interpreter_call", |
| 769 | id=state.current_item_id, |
| 770 | code=previous_item.content[0].text, |
| 771 | container_id="auto", |
| 772 | outputs=[], |
| 773 | status="completed", |
| 774 | ), |
| 775 | ) |
| 776 | ) |
| 777 | return events |
| 778 | |
| 779 | |
| 780 | def emit_tool_action_events( |
| 781 | previous_item: HarmonyMessage, |
| 782 | state: StreamingState, |
| 783 | tool_server: ToolServer | None, |
| 784 | ) -> list[StreamingResponsesResponse]: |
| 785 | """Emit events for a completed assistant action turn.""" |
| 786 | # Handle browser tool |
| 787 | if ( |
| 788 | previous_item.author.role == "assistant" |
| 789 | and previous_item.recipient is not None |
| 790 | and previous_item.recipient.startswith("browser.") |
| 791 | and tool_server is not None |
| 792 | and tool_server.has_tool("browser") |
| 793 | ): |
| 794 | return emit_browser_tool_events(previous_item, state) |
| 795 | |
| 796 | return [] |
| 797 | |
| 798 | |
| 799 | # ===================================================================== |
| 800 | # Simple streaming helpers |
| 801 | # ===================================================================== |
| 802 | |
| 803 | |
| 804 | class _StateType(Enum): |
| 805 | NONE = auto() |
| 806 | CONTENT = auto() |
| 807 | REASONING = auto() |
| 808 | TOOL_CALL = auto() |
| 809 | |
| 810 | |
| 811 | @dataclass |
| 812 | class SimpleStreamingState: |
| 813 | output_index: int = 0 |
| 814 | current_item_id: str = "" |
| 815 | content_index: int = 0 |
| 816 | accumulated_text: str = "" |
| 817 | accumulated_logprobs: list[response_output_text.Logprob] = field( |
| 818 | default_factory=list |
| 819 | ) |
| 820 | tool_call_id: str = "" |
| 821 | tool_call_name: str = "" |
| 822 | tool_call_namespace: str | None = None |
| 823 | tool_call_index: int | None = None |
| 824 | current_state: _StateType = field(default_factory=lambda: _StateType.NONE) |
| 825 | |
| 826 | |
| 827 | def emit_simple_content_open( |
| 828 | state: SimpleStreamingState, |
| 829 | ) -> list[StreamingResponsesResponse]: |
| 830 | state.current_state = _StateType.CONTENT |
| 831 | state.current_item_id = f"msg_{random_uuid()}" |
| 832 | state.content_index = 0 |
| 833 | state.accumulated_text = "" |
| 834 | state.accumulated_logprobs = [] |
| 835 | return [ |
| 836 | ResponseOutputItemAddedEvent( |
| 837 | type="response.output_item.added", |
| 838 | sequence_number=-1, |
| 839 | output_index=state.output_index, |
| 840 | item=ResponseOutputMessage( |
| 841 | id=state.current_item_id, |
| 842 | type="message", |
| 843 | role="assistant", |
| 844 | content=[], |
| 845 | status="in_progress", |
| 846 | ), |
| 847 | ), |
| 848 | ResponseContentPartAddedEvent( |
| 849 | type="response.content_part.added", |
| 850 | sequence_number=-1, |
| 851 | output_index=state.output_index, |
| 852 | item_id=state.current_item_id, |
| 853 | content_index=state.content_index, |
| 854 | part=ResponseOutputText( |
| 855 | type="output_text", |
| 856 | text="", |
| 857 | annotations=[], |
| 858 | logprobs=[], |
| 859 | ), |
| 860 | ), |
| 861 | ] |
| 862 | |
| 863 | |
| 864 | def _to_stream_logprobs( |
| 865 | logprobs: list[response_output_text.Logprob], |
| 866 | ) -> list[response_text_delta_event.Logprob]: |
| 867 | return [ |
| 868 | response_text_delta_event.Logprob( |
| 869 | token=lp.token, |
| 870 | logprob=lp.logprob, |
| 871 | top_logprobs=[ |
| 872 | response_text_delta_event.LogprobTopLogprob( |
| 873 | token=tl.token, logprob=tl.logprob |
| 874 | ) |
| 875 | for tl in lp.top_logprobs |
| 876 | ], |
| 877 | ) |
| 878 | for lp in logprobs |
| 879 | ] |
| 880 | |
| 881 | |
| 882 | def emit_simple_content_delta( |
| 883 | state: SimpleStreamingState, |
| 884 | delta: str, |
| 885 | logprobs: list[response_output_text.Logprob] | None = None, |
| 886 | ) -> list[StreamingResponsesResponse]: |
| 887 | state.accumulated_text += delta |
| 888 | if logprobs: |
| 889 | state.accumulated_logprobs.extend(logprobs) |
| 890 | return [ |
| 891 | ResponseTextDeltaEvent( |
| 892 | type="response.output_text.delta", |
| 893 | sequence_number=-1, |
| 894 | content_index=state.content_index, |
| 895 | output_index=state.output_index, |
| 896 | item_id=state.current_item_id, |
| 897 | delta=delta, |
| 898 | logprobs=_to_stream_logprobs(logprobs or []), |
| 899 | ) |
| 900 | ] |
| 901 | |
| 902 | |
| 903 | def emit_simple_content_done( |
| 904 | state: SimpleStreamingState, |
| 905 | ) -> list[StreamingResponsesResponse]: |
| 906 | part = ResponseOutputText( |
| 907 | type="output_text", |
| 908 | text=state.accumulated_text, |
| 909 | annotations=[], |
| 910 | logprobs=state.accumulated_logprobs or None, |
| 911 | ) |
| 912 | events: list[StreamingResponsesResponse] = [ |
| 913 | ResponseTextDoneEvent( |
| 914 | type="response.output_text.done", |
| 915 | sequence_number=-1, |
| 916 | output_index=state.output_index, |
| 917 | content_index=state.content_index, |
| 918 | text=state.accumulated_text, |
| 919 | logprobs=[], |
| 920 | item_id=state.current_item_id, |
| 921 | ), |
| 922 | ResponseContentPartDoneEvent( |
| 923 | type="response.content_part.done", |
| 924 | sequence_number=-1, |
| 925 | item_id=state.current_item_id, |
| 926 | output_index=state.output_index, |
| 927 | content_index=state.content_index, |
| 928 | part=part, |
| 929 | ), |
| 930 | ResponseOutputItemDoneEvent( |
| 931 | type="response.output_item.done", |
| 932 | sequence_number=-1, |
| 933 | output_index=state.output_index, |
| 934 | item=ResponseOutputMessage( |
| 935 | id=state.current_item_id, |
| 936 | type="message", |
| 937 | role="assistant", |
| 938 | content=[part] if state.accumulated_text else [], |
| 939 | status="completed", |
| 940 | ), |
| 941 | ), |
| 942 | ] |
| 943 | state.output_index += 1 |
| 944 | state.current_state = _StateType.NONE |
| 945 | return events |
| 946 | |
| 947 | |
| 948 | def emit_simple_reasoning_open( |
| 949 | state: SimpleStreamingState, |
| 950 | ) -> list[StreamingResponsesResponse]: |
| 951 | state.current_state = _StateType.REASONING |
| 952 | state.current_item_id = f"rs_{random_uuid()}" |
| 953 | state.content_index = 0 |
| 954 | state.accumulated_text = "" |
| 955 | return [ |
| 956 | ResponseOutputItemAddedEvent( |
| 957 | type="response.output_item.added", |
| 958 | sequence_number=-1, |
| 959 | output_index=state.output_index, |
| 960 | item=ResponseReasoningItem( |
| 961 | type="reasoning", |
| 962 | id=state.current_item_id, |
| 963 | summary=[], |
| 964 | status="in_progress", |
| 965 | ), |
| 966 | ), |
| 967 | ResponseContentPartAddedEvent( |
| 968 | type="response.content_part.added", |
| 969 | sequence_number=-1, |
| 970 | output_index=state.output_index, |
| 971 | item_id=state.current_item_id, |
| 972 | content_index=state.content_index, |
| 973 | part={"text": "", "type": "reasoning_text"}, |
| 974 | ), |
| 975 | ] |
| 976 | |
| 977 | |
| 978 | def emit_simple_reasoning_delta( |
| 979 | state: SimpleStreamingState, |
| 980 | delta: str, |
| 981 | ) -> list[StreamingResponsesResponse]: |
| 982 | state.accumulated_text += delta |
| 983 | return [ |
| 984 | ResponseReasoningTextDeltaEvent( |
| 985 | type="response.reasoning_text.delta", |
| 986 | item_id=state.current_item_id, |
| 987 | sequence_number=-1, |
| 988 | output_index=state.output_index, |
| 989 | content_index=state.content_index, |
| 990 | delta=delta, |
| 991 | ) |
| 992 | ] |
| 993 | |
| 994 | |
| 995 | def emit_simple_reasoning_done( |
| 996 | state: SimpleStreamingState, |
| 997 | ) -> list[StreamingResponsesResponse]: |
| 998 | part = ResponseReasoningTextContent( |
| 999 | text=state.accumulated_text, |
| 1000 | type="reasoning_text", |
| 1001 | ) |
| 1002 | events: list[StreamingResponsesResponse] = [ |
| 1003 | ResponseReasoningTextDoneEvent( |
| 1004 | type="response.reasoning_text.done", |
| 1005 | item_id=state.current_item_id, |
| 1006 | sequence_number=-1, |
| 1007 | output_index=state.output_index, |
| 1008 | content_index=state.content_index, |
| 1009 | text=state.accumulated_text, |
| 1010 | ), |
| 1011 | ResponseContentPartDoneEvent( |
| 1012 | type="response.content_part.done", |
| 1013 | sequence_number=-1, |
| 1014 | item_id=state.current_item_id, |
| 1015 | output_index=state.output_index, |
| 1016 | content_index=state.content_index, |
| 1017 | part=part.model_dump(), |
| 1018 | ), |
| 1019 | ResponseOutputItemDoneEvent( |
| 1020 | type="response.output_item.done", |
| 1021 | sequence_number=-1, |
| 1022 | output_index=state.output_index, |
| 1023 | item=ResponseReasoningItem( |
| 1024 | type="reasoning", |
| 1025 | content=[part], |
| 1026 | status="completed", |
| 1027 | id=state.current_item_id, |
| 1028 | summary=[], |
| 1029 | ), |
| 1030 | ), |
| 1031 | ] |
| 1032 | state.output_index += 1 |
| 1033 | state.current_state = _StateType.NONE |
| 1034 | return events |
| 1035 | |
| 1036 | |
| 1037 | def emit_simple_tool_call_open( |
| 1038 | state: SimpleStreamingState, |
| 1039 | name: str, |
| 1040 | index: int | None, |
| 1041 | namespace: str | None = None, |
| 1042 | call_id: str | None = None, |
| 1043 | ) -> list[StreamingResponsesResponse]: |
| 1044 | state.current_state = _StateType.TOOL_CALL |
| 1045 | state.current_item_id = f"fc_{random_uuid()}" |
| 1046 | state.tool_call_id = call_id or make_tool_call_id() |
| 1047 | state.tool_call_name = name |
| 1048 | state.tool_call_namespace = namespace |
| 1049 | state.tool_call_index = index |
| 1050 | state.accumulated_text = "" |
| 1051 | return [ |
| 1052 | ResponseOutputItemAddedEvent( |
| 1053 | type="response.output_item.added", |
| 1054 | sequence_number=-1, |
| 1055 | output_index=state.output_index, |
| 1056 | item=ResponseFunctionToolCallItem( |
| 1057 | type="function_call", |
| 1058 | id=state.current_item_id, |
| 1059 | call_id=state.tool_call_id, |
| 1060 | name=name, |
| 1061 | namespace=namespace, |
| 1062 | arguments="", |
| 1063 | status="in_progress", |
| 1064 | ), |
| 1065 | ), |
| 1066 | ] |
| 1067 | |
| 1068 | |
| 1069 | def emit_simple_tool_call_delta( |
| 1070 | state: SimpleStreamingState, |
| 1071 | delta: str, |
| 1072 | ) -> list[StreamingResponsesResponse]: |
| 1073 | state.accumulated_text += delta |
| 1074 | return [ |
| 1075 | ResponseFunctionCallArgumentsDeltaEvent( |
| 1076 | type="response.function_call_arguments.delta", |
| 1077 | sequence_number=-1, |
| 1078 | output_index=state.output_index, |
| 1079 | item_id=state.current_item_id, |
| 1080 | delta=delta, |
| 1081 | ) |
| 1082 | ] |
| 1083 | |
| 1084 | |
| 1085 | def emit_simple_tool_call_done( |
| 1086 | state: SimpleStreamingState, |
| 1087 | ) -> list[StreamingResponsesResponse]: |
| 1088 | events: list[StreamingResponsesResponse] = [ |
| 1089 | ResponseFunctionCallArgumentsDoneEvent( |
| 1090 | type="response.function_call_arguments.done", |
| 1091 | sequence_number=-1, |
| 1092 | output_index=state.output_index, |
| 1093 | item_id=state.current_item_id, |
| 1094 | arguments=state.accumulated_text, |
| 1095 | name=state.tool_call_name, |
| 1096 | ), |
| 1097 | ResponseOutputItemDoneEvent( |
| 1098 | type="response.output_item.done", |
| 1099 | sequence_number=-1, |
| 1100 | output_index=state.output_index, |
| 1101 | item=ResponseFunctionToolCall( |
| 1102 | type="function_call", |
| 1103 | name=state.tool_call_name, |
| 1104 | namespace=state.tool_call_namespace, |
| 1105 | arguments=state.accumulated_text, |
| 1106 | status="completed", |
| 1107 | id=state.current_item_id, |
| 1108 | call_id=state.tool_call_id, |
| 1109 | ), |
| 1110 | ), |
| 1111 | ] |
| 1112 | state.output_index += 1 |
| 1113 | state.tool_call_namespace = None |
| 1114 | state.current_state = _StateType.NONE |
| 1115 | return events |
| 1116 | |
| 1117 | |
| 1118 | class _StateHandlers(NamedTuple): |
| 1119 | """Tuple for each state: open(start), delta(chunk), done(finish).""" |
| 1120 | |
| 1121 | open_fn: Callable[..., list[StreamingResponsesResponse]] |
| 1122 | delta_fn: Callable[..., list[StreamingResponsesResponse]] |
| 1123 | done_fn: Callable[..., list[StreamingResponsesResponse]] |
| 1124 | |
| 1125 | |
| 1126 | def split_delta(delta: DeltaMessage) -> list[DeltaMessage]: |
| 1127 | """Decompose a DeltaMessage with multiple fields into atomic deltas. |
| 1128 | |
| 1129 | The Responses API emits typed SSE events (one type per event), so a |
| 1130 | compound DeltaMessage must be split before entering the state machine. |
| 1131 | Order: reasoning -> content -> tool_calls (grouped by index). |
| 1132 | """ |
| 1133 | has_reasoning = delta.reasoning is not None |
| 1134 | has_content = delta.content is not None |
| 1135 | has_tools = bool(delta.tool_calls) |
| 1136 | parts = int(has_reasoning) + int(has_content) + int(has_tools) |
| 1137 | |
| 1138 | if parts <= 1 and ( |
| 1139 | not has_tools |
| 1140 | or len({tc.index for tc in delta.tool_calls if tc.index is not None}) <= 1 |
| 1141 | ): |
| 1142 | return [delta] |
| 1143 | |
| 1144 | deltas: list[DeltaMessage] = [] |
| 1145 | if has_reasoning: |
| 1146 | deltas.append(DeltaMessage(reasoning=delta.reasoning)) |
| 1147 | if has_content: |
| 1148 | deltas.append(DeltaMessage(content=delta.content)) |
| 1149 | if has_tools: |
| 1150 | groups: dict[int | None, list[DeltaToolCall]] = {} |
| 1151 | for tc in delta.tool_calls: |
| 1152 | groups.setdefault(tc.index, []).append(tc) |
| 1153 | for tcs in groups.values(): |
| 1154 | deltas.append(DeltaMessage(tool_calls=tcs)) |
| 1155 | return deltas or [delta] |
| 1156 | |
| 1157 | |
| 1158 | class SimpleStreamingEventProcessor: |
| 1159 | """State-machine processor for the simple (non-Harmony) streaming path. |
| 1160 | |
| 1161 | Core flow: |
| 1162 | 1. Resolve the target state from the delta_message |
| 1163 | (CONTENT / REASONING / TOOL_CALL). |
| 1164 | 2. If the target state differs from the current one, |
| 1165 | close_current() then open() the new state. |
| 1166 | 3. emit_delta() produces the incremental events for the state. |
| 1167 | |
| 1168 | State lifecycle: |
| 1169 | open() -> repeated emit_delta() -> close_current() |
| 1170 | """ |
| 1171 | |
| 1172 | _STATE_HANDLERS: ClassVar[dict[_StateType, _StateHandlers]] = { |
| 1173 | _StateType.CONTENT: _StateHandlers( |
| 1174 | emit_simple_content_open, |
| 1175 | emit_simple_content_delta, |
| 1176 | emit_simple_content_done, |
| 1177 | ), |
| 1178 | _StateType.REASONING: _StateHandlers( |
| 1179 | emit_simple_reasoning_open, |
| 1180 | emit_simple_reasoning_delta, |
| 1181 | emit_simple_reasoning_done, |
| 1182 | ), |
| 1183 | _StateType.TOOL_CALL: _StateHandlers( |
| 1184 | emit_simple_tool_call_open, |
| 1185 | emit_simple_tool_call_delta, |
| 1186 | emit_simple_tool_call_done, |
| 1187 | ), |
| 1188 | } |
| 1189 | |
| 1190 | def __init__( |
| 1191 | self, |
| 1192 | state: SimpleStreamingState | None = None, |
| 1193 | tools: list[Tool] | None = None, |
| 1194 | ) -> None: |
| 1195 | self.state = state or SimpleStreamingState() |
| 1196 | self.tool_call_name_map = build_responses_tool_call_name_map(tools) |
| 1197 | self.output_items: list[ResponseOutputItem] = [] |
| 1198 | |
| 1199 | def resolve_target_state( |
| 1200 | self, delta_message: DeltaMessage |
| 1201 | ) -> tuple[_StateType, Any]: |
| 1202 | """Decide which state the next delta belongs to. |
| 1203 | |
| 1204 | Priority: TOOL_CALL > REASONING > CONTENT, fallback to NONE. |
| 1205 | For TOOL_CALL the first tool_call object is also returned so |
| 1206 | callers can detect a switch between consecutive tools. |
| 1207 | """ |
| 1208 | if ( |
| 1209 | delta_message.tool_calls |
| 1210 | and delta_message.tool_calls[0].function is not None |
| 1211 | ): |
| 1212 | return _StateType.TOOL_CALL, delta_message.tool_calls[0] |
| 1213 | if delta_message.reasoning is not None: |
| 1214 | return _StateType.REASONING, None |
| 1215 | if delta_message.content: |
| 1216 | return _StateType.CONTENT, None |
| 1217 | return _StateType.NONE, None |
| 1218 | |
| 1219 | def needs_transition(self, target_state: _StateType, tool_call: Any) -> bool: |
| 1220 | """Return True when we must close the current state and open a new one. |
| 1221 | |
| 1222 | Two cases trigger a transition: |
| 1223 | 1. The target state differs from the current state |
| 1224 | (e.g. CONTENT -> TOOL_CALL). |
| 1225 | 2. We are already in TOOL_CALL but the next tool_call has a |
| 1226 | different index (multiple consecutive tool calls). |
| 1227 | """ |
| 1228 | if self.state.current_state != target_state: |
| 1229 | return True |
| 1230 | return ( |
| 1231 | target_state == _StateType.TOOL_CALL |
| 1232 | and tool_call is not None |
| 1233 | and self.state.tool_call_index is not None |
| 1234 | and tool_call.index is not None |
| 1235 | and self.state.tool_call_index != tool_call.index |
| 1236 | ) |
| 1237 | |
| 1238 | def close_current(self) -> list[StreamingResponsesResponse]: |
| 1239 | """Close the current state and emit its 'done' event sequence.""" |
| 1240 | handlers = self._STATE_HANDLERS.get(self.state.current_state) |
| 1241 | if handlers is None: |
| 1242 | return [] |
| 1243 | events = handlers.done_fn(self.state) |
| 1244 | self.output_items.extend( |
| 1245 | event.item |
| 1246 | for event in events |
| 1247 | if isinstance(event, ResponseOutputItemDoneEvent) |
| 1248 | ) |
| 1249 | return events |
| 1250 | |
| 1251 | def open( |
| 1252 | self, target_state: _StateType, tool_call: Any = None |
| 1253 | ) -> list[StreamingResponsesResponse]: |
| 1254 | """Open a new state and emit its 'added' / 'open' event sequence.""" |
| 1255 | handlers = self._STATE_HANDLERS[target_state] |
| 1256 | if target_state == _StateType.TOOL_CALL: |
| 1257 | assert tool_call is not None |
| 1258 | call_name = resolve_responses_tool_call_name( |
| 1259 | tool_call.function.name, |
| 1260 | tool_call_name_map=self.tool_call_name_map, |
| 1261 | ) |
| 1262 | return handlers.open_fn( |
| 1263 | self.state, |
| 1264 | call_name.name, |
| 1265 | tool_call.index, |
| 1266 | call_name.namespace, |
| 1267 | tool_call.id, |
| 1268 | ) |
| 1269 | return handlers.open_fn(self.state) |
| 1270 | |
| 1271 | def emit_delta( |
| 1272 | self, |
| 1273 | delta_message: DeltaMessage, |
| 1274 | output: CompletionOutput, |
| 1275 | get_logprobs: Callable[[CompletionOutput], list[response_output_text.Logprob]] |
| 1276 | | None = None, |
| 1277 | ) -> list[StreamingResponsesResponse]: |
| 1278 | """Emit incremental events for the current state from the delta.""" |
| 1279 | handlers = self._STATE_HANDLERS[self.state.current_state] |
| 1280 | |
| 1281 | if self.state.current_state == _StateType.TOOL_CALL: |
| 1282 | assert delta_message.tool_calls is not None |
| 1283 | combined_args = "" |
| 1284 | for tc in delta_message.tool_calls: |
| 1285 | if tc.function is not None and tc.function.arguments: |
| 1286 | combined_args += tc.function.arguments |
| 1287 | if combined_args: |
| 1288 | return handlers.delta_fn(self.state, combined_args) |
| 1289 | return [] |
| 1290 | elif self.state.current_state == _StateType.REASONING: |
| 1291 | assert delta_message.reasoning is not None |
| 1292 | return handlers.delta_fn(self.state, delta_message.reasoning) |
| 1293 | elif self.state.current_state == _StateType.CONTENT: |
| 1294 | assert delta_message.content is not None |
| 1295 | logprobs = get_logprobs(output) if get_logprobs else [] |
| 1296 | return handlers.delta_fn(self.state, delta_message.content, logprobs) |
| 1297 | return [] |