tap_output_aiter() — langchain Function Reference

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

Function python Observability Tracers calls 1

Entity Profile

Observability→ Tracers→ tap_output_aiter() — langchain Function Reference

Dependency Diagram

graph TD
  c52d25da_e50c_d411_085c_c87f3733b1c6["tap_output_aiter()"]
  78fc514f_5439_00c6_7e00_021af9ce4e07["LogStreamCallbackHandler"]
  c52d25da_e50c_d411_085c_c87f3733b1c6 -->|defined in| 78fc514f_5439_00c6_7e00_021af9ce4e07
  c5f1fbd2_23b6_75cb_1f29_b9e028c332a0["send()"]
  c52d25da_e50c_d411_085c_c87f3733b1c6 -->|calls| c5f1fbd2_23b6_75cb_1f29_b9e028c332a0
  style c52d25da_e50c_d411_085c_c87f3733b1c6 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/tracers/log_stream.py lines 325–356

    async def tap_output_aiter(
        self, run_id: UUID, output: AsyncIterator[T]
    ) -> AsyncIterator[T]:
        """Tap an output async iterator to stream its values to the log.

        Args:
            run_id: The ID of the run.
            output: The output async iterator.

        Yields:
            The output value.
        """
        async for chunk in output:
            # root run is handled in .astream_log()
            # if we can't find the run silently ignore
            # eg. because this run wasn't included in the log
            if (
                run_id != self.root_id
                and (key := self._key_map_by_run_id.get(run_id))
                and (
                    not self.send(
                        {
                            "op": "add",
                            "path": f"/logs/{key}/streamed_output/-",
                            "value": chunk,
                        }
                    )
                )
            ):
                break

            yield chunk