Turn Loop 与 Streaming Round 深度解析

  67 def latest_user_text(messages: list[dict[str, Any]], host_texts: Collection[str] = ()) -> str:
  68     """The most recent user-authored text; "" when the conversation has none."""
  69     for message in reversed(messages):
  70         if message.get("role") != "user":
  71             continue
  72         content = message.get("content")
  73         texts = _text_blocks(content)
  74         if texts and all(text in host_texts for text in texts):
  75             continue
  76         if texts:
  77             return "\n".join(texts)
  78         # A message of tool results is the loop's own; anything else ends the search.
  79         if isinstance(content, list) and any(
  80             isinstance(block, dict) and block.get("type") == "tool_result" for block in content
  81         ):
  82             continue
  83         return ""
  84     return ""
  85 
  86 
  87 def latest_exchange(
  88     messages: list[dict[str, Any]], host_texts: Collection[str] = ()
  89 ) -> list[dict[str, Any]]:
  90     """The slice from the most recent user-authored message onward: what one turn's
  91     extraction pass reads, so earlier turns are not re-extracted in new words."""
  92     for index in range(len(messages) - 1, -1, -1):
  93         if _is_user_text(messages[index], host_texts):
  94             return messages[index:]
  95     return messages

 174 class EagerDispatcher:
 175     """Tool execution started mid-stream. When a ``tool_use`` block's buffered JSON
 176     parses at ``content_block_stop``, the call's arguments are final — the loop starts
 177     executing then, while the model writes the rest of the round, so tool time overlaps
 178     generation time. ``collect`` is the post-stream join: started calls are awaited,
 179     anything the stream path missed (unparseable buffer, dispatch off) executes from the
 180     final message's canonical args, and every call runs exactly once. ``cancel`` is the
 181     error and malformed-stream backstop; a started task must not outlive its turn."""
 182 
 183     def __init__(
 184         self, execute: Callable[[str, dict[str, Any]], Awaitable[ToolOutcome]], enabled: bool
 185     ) -> None:
 186         self._execute = execute
 187         self._enabled = enabled
 188         self._tasks: dict[str, asyncio.Future[ToolOutcome]] = {}
 189 
 190     def started(self, tool_use_id: str) -> bool:
 191         return tool_use_id in self._tasks
 192 
 193     def dispatch(self, name: str, tool_use_id: str, args: dict[str, Any] | None) -> bool:
 194         """Start ``name`` on ``args``; False when it did not start (dispatch off, already
 195         started, or no parsed args) and the join executes it later."""
 196         if not self._enabled or args is None or tool_use_id in self._tasks:
 197             return False
 198         self._tasks[tool_use_id] = asyncio.ensure_future(self._execute(name, args))
 199         return True
 200 
 201     def settle(self, tool_use_id: str, outcome: ToolOutcome) -> None:
 202         """Give ``tool_use_id`` its outcome without executing it (a call whose input
 203         never parsed); the join returns it in place. A started call keeps its task."""
 204         if tool_use_id in self._tasks:
 205             return
 206         settled: asyncio.Future[ToolOutcome] = asyncio.get_running_loop().create_future()
 207         settled.set_result(outcome)
 208         self._tasks[tool_use_id] = settled
 209 
 210     async def collect(self, tool_uses: list[Any]) -> list[ToolOutcome]:
 211         return await asyncio.gather(
 212             *(
 213                 self._tasks[block.id]
 214                 if block.id in self._tasks
 215                 else self._execute(block.name, dict(block.input or {}))
 216                 for block in tool_uses
 217             )
 218         )
 219 
 220     def cancel(self) -> None:

 401 def salvage_round(
 402     streamed: StreamedRound,
 403     dispatcher: EagerDispatcher,
 404     caller: logging.Logger,
 405     session_id: str | None,
 406     round_index: int,
 407 ) -> tuple[dict[str, Any] | None, list[Any], set[str]]:
 408     """:meth:`StreamedRound.salvaged` for a round the SDK's accumulator abandoned, with
 409     each unreadable client call settled on ``dispatcher`` as an error result so the model
 410     sends it again, and a WARNING per unreadable call on ``caller`` naming it and the size
 411     of what arrived, never the input."""
 412     reply, tool_uses, unreadable = streamed.salvaged()
 413     for tool in streamed.tools.values():
 414         if tool.id not in unreadable:
 415             continue
 416         caller.warning(
 417             "tool input unreadable session=%s round=%d tool=%s id=%s chars=%d",
 418             session_tag(session_id),
 419             round_index,
 420             tool.name,
 421             tool.id,
 422             len(tool.buffer),
 423             stacklevel=2,
 424         )
 425         if not tool.server:
 426             dispatcher.settle(tool.id, ToolOutcome.error(UNREADABLE_INPUT_TEXT))
 427     return reply, tool_uses, unreadable