Represents an in-memory paragraph buffer for streamed chunks. A BufferEntry collects sequential message chunks belonging to the same logical paragraph. It maintains a stable `item_id` so streamed chunks can be correlated with the final persisted ConversationItem.
| 37 | |
| 38 | |
| 39 | class BufferEntry: |
| 40 | """Represents an in-memory paragraph buffer for streamed chunks. |
| 41 | |
| 42 | A BufferEntry collects sequential message chunks belonging to the same |
| 43 | logical paragraph. It maintains a stable `item_id` so streamed chunks can |
| 44 | be correlated with the final persisted ConversationItem. |
| 45 | """ |
| 46 | |
| 47 | def __init__( |
| 48 | self, |
| 49 | item_id: Optional[str] = None, |
| 50 | role: Optional[Role] = None, |
| 51 | agent_name: Optional[str] = None, |
| 52 | ): |
| 53 | self.parts: List[str] = [] |
| 54 | self.last_updated: float = time.monotonic() |
| 55 | # Stable paragraph id for this buffer entry. Reused across streamed chunks |
| 56 | # until this entry is flushed (debounce/boundary). On size-based flush, |
| 57 | # we rotate to a new paragraph id for subsequent chunks. |
| 58 | self.item_id: str = item_id or generate_item_id() |
| 59 | self.role: Optional[Role] = role |
| 60 | self.agent_name: Optional[str] = agent_name |
| 61 | |
| 62 | def append(self, text: str): |
| 63 | """Append a chunk of text to this buffer and update the timestamp.""" |
| 64 | if text: |
| 65 | self.parts.append(text) |
| 66 | self.last_updated = time.monotonic() |
| 67 | |
| 68 | def snapshot_payload(self) -> Optional[BaseResponseDataPayload]: |
| 69 | """Return the current aggregate content as a payload without clearing. |
| 70 | |
| 71 | Returns None when there is no content buffered. |
| 72 | """ |
| 73 | if not self.parts: |
| 74 | return None |
| 75 | content = "".join(self.parts) |
| 76 | return BaseResponseDataPayload(content=content) |
| 77 | |
| 78 | |
| 79 | class ResponseBuffer: |
no outgoing calls