on_custom_event() — langchain Function Reference

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

Function python Observability Tracers calls 2

Entity Profile

Observability→ Tracers→ on_custom_event() — langchain Function Reference

Dependency Diagram

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

Relationship Graph

Source Code

libs/core/langchain_core/tracers/event_stream.py lines 405–425

    async def on_custom_event(
        self,
        name: str,
        data: Any,
        *,
        run_id: UUID,
        tags: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        """Generate a custom astream event."""
        event = CustomStreamEvent(
            event="on_custom_event",
            run_id=str(run_id),
            name=name,
            tags=tags or [],
            metadata=metadata or {},
            data=data,
            parent_ids=self._get_parent_ids(run_id),
        )
        self._send(event, name)