test_sync_callback_manager() — langchain Function Reference

Architecture documentation for the test_sync_callback_manager() function in test_dispatch_custom_event.py from the langchain codebase.

Function python CoreAbstractions RunnableInterface calls 1

Entity Profile

CoreAbstractions→ RunnableInterface→ test_sync_callback_manager() — langchain Function Reference

Dependency Diagram

graph TD
  a434e945_618d_6170_58c9_07fa320dce77["test_sync_callback_manager()"]
  f4cde4a2_4706_d7d1_26f9_82085d97a270["test_dispatch_custom_event.py"]
  a434e945_618d_6170_58c9_07fa320dce77 -->|defined in| f4cde4a2_4706_d7d1_26f9_82085d97a270
  505e68ca_8d30_9fd9_7a8b_6412fbd762b6["on_custom_event()"]
  a434e945_618d_6170_58c9_07fa320dce77 -->|calls| 505e68ca_8d30_9fd9_7a8b_6412fbd762b6
  style a434e945_618d_6170_58c9_07fa320dce77 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/callbacks/test_dispatch_custom_event.py lines 114–157

def test_sync_callback_manager() -> None:
    """Test async callback manager."""

    class CustomCallbackManager(BaseCallbackHandler):
        def __init__(self) -> None:
            self.events: list[Any] = []

        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:
            assert kwargs == {}
            self.events.append(
                (
                    name,
                    data,
                    run_id,
                    tags,
                    metadata,
                )
            )

    callback = CustomCallbackManager()

    run_id = uuid.UUID(int=7)

    @RunnableLambda
    def foo(x: int, config: RunnableConfig) -> int:
        dispatch_custom_event("event1", {"x": x})
        dispatch_custom_event("event2", {"x": x}, config=config)
        return x

    foo.invoke(1, {"callbacks": [callback], "run_id": run_id})

    assert callback.events == [
        ("event1", {"x": 1}, UUID("00000000-0000-0000-0000-000000000007"), [], {}),
        ("event2", {"x": 1}, UUID("00000000-0000-0000-0000-000000000007"), [], {}),
    ]