tap_output_aiter() — langchain Function Reference

Architecture documentation for the tap_output_aiter() function in event_stream.py from the langchain codebase.

Function python Observability Tracers calls 2 called by 1

Entity Profile

Observability→ Tracers→ tap_output_aiter() — langchain Function Reference

Dependency Diagram

graph TD
  1b967044_ad1d_2158_3ce0_f8f5637ebfeb["tap_output_aiter()"]
  33d093c4_1ed0_fc6a_17c6_762d4c5cfa04["_AstreamEventsCallbackHandler"]
  1b967044_ad1d_2158_3ce0_f8f5637ebfeb -->|defined in| 33d093c4_1ed0_fc6a_17c6_762d4c5cfa04
  96f18cbb_b657_2936_be82_6eeac3f72e0a["_astream_events_implementation_v2()"]
  96f18cbb_b657_2936_be82_6eeac3f72e0a -->|calls| 1b967044_ad1d_2158_3ce0_f8f5637ebfeb
  9aed8e4f_9d4c_016f_aa43_c5908015cf8d["_get_parent_ids()"]
  1b967044_ad1d_2158_3ce0_f8f5637ebfeb -->|calls| 9aed8e4f_9d4c_016f_aa43_c5908015cf8d
  87f79bee_f9c5_8262_1829_62f633c4f870["_send()"]
  1b967044_ad1d_2158_3ce0_f8f5637ebfeb -->|calls| 87f79bee_f9c5_8262_1829_62f633c4f870
  style 1b967044_ad1d_2158_3ce0_f8f5637ebfeb fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/tracers/event_stream.py lines 180–233

    async def tap_output_aiter(
        self, run_id: UUID, output: AsyncIterator[T]
    ) -> AsyncIterator[T]:
        """Tap the output aiter.

        This method is used to tap the output of a `Runnable` that produces an async
        iterator. It is used to generate stream events for the output of the `Runnable`.

        Args:
            run_id: The ID of the run.
            output: The output of the `Runnable`.

        Yields:
            The output of the `Runnable`.
        """
        sentinel = object()
        # atomic check and set
        tap = self.is_tapped.setdefault(run_id, sentinel)
        # wait for first chunk
        first = await anext(output, sentinel)
        if first is sentinel:
            return
        # get run info
        run_info = self.run_map.get(run_id)
        if run_info is None:
            # run has finished, don't issue any stream events
            yield cast("T", first)
            return
        if tap is sentinel:
            # if we are the first to tap, issue stream events
            event: StandardStreamEvent = {
                "event": f"on_{run_info['run_type']}_stream",
                "run_id": str(run_id),
                "name": run_info["name"],
                "tags": run_info["tags"],
                "metadata": run_info["metadata"],
                "data": {},
                "parent_ids": self._get_parent_ids(run_id),
            }
            self._send({**event, "data": {"chunk": first}}, run_info["run_type"])
            yield cast("T", first)
            # consume the rest of the output
            async for chunk in output:
                self._send(
                    {**event, "data": {"chunk": chunk}},
                    run_info["run_type"],
                )
                yield chunk
        else:
            # otherwise just pass through
            yield cast("T", first)
            # consume the rest of the output
            async for chunk in output:
                yield chunk