Compact structural skeleton of a spliced stream — no text content.
(events: list[BetaRawMessageStreamEvent])
| 169 | |
| 170 | |
| 171 | def skeleton(events: list[BetaRawMessageStreamEvent]) -> list[str]: |
| 172 | """Compact structural skeleton of a spliced stream — no text content.""" |
| 173 | out: list[str] = [] |
| 174 | for event in events: |
| 175 | if event.type == "content_block_start": |
| 176 | block = event.content_block |
| 177 | if block.type == "fallback": |
| 178 | label = f"fallback{{{block.from_.model}->{block.to.model}}}" |
| 179 | else: |
| 180 | label = block.type |
| 181 | out.append(f"start[{event.index}] {label}") |
| 182 | elif event.type == "content_block_delta": |
| 183 | out.append(f"delta[{event.index}] {event.delta.type}") |
| 184 | elif event.type == "content_block_stop": |
| 185 | out.append(f"stop[{event.index}]") |
| 186 | elif event.type == "message_delta": |
| 187 | iterations = ",".join(f"{i.type}:{i.model}" for i in (event.usage.iterations or [])) # type: ignore[union-attr] |
| 188 | out.append(f"message_delta {event.delta.stop_reason} iter=[{iterations}]") |
| 189 | else: |
| 190 | out.append(event.type) |
| 191 | return out |
| 192 | |
| 193 | |
| 194 | def block_starts(events: list[BetaRawMessageStreamEvent]) -> list[tuple[int, str]]: |
no outgoing calls
no test coverage detected