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