tap_output_iter() — langchain Function Reference

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

Function python Observability Tracers calls 2

Entity Profile

Observability→ Tracers→ tap_output_iter() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/langchain_core/tracers/event_stream.py lines 235–283

    def tap_output_iter(self, run_id: UUID, output: Iterator[T]) -> Iterator[T]:
        """Tap the output iter.

        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 = next(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
            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
            for chunk in output:
                yield chunk