test_with_explicit_config() — langchain Function Reference

Architecture documentation for the test_with_explicit_config() function in test_runnable_events_v2.py from the langchain codebase.

Function python LangChainCore Runnables calls 1

Entity Profile

LangChainCore→ Runnables→ test_with_explicit_config() — langchain Function Reference

Dependency Diagram

graph TD
  b9b27172_6269_f5cd_d81e_b606b7c2f1c7["test_with_explicit_config()"]
  33c02978_2077_5819_7048_bc2a81e80625["test_runnable_events_v2.py"]
  b9b27172_6269_f5cd_d81e_b606b7c2f1c7 -->|defined in| 33c02978_2077_5819_7048_bc2a81e80625
  716d2a5e_dc8e_3cae_e044_b56b06bee655["_collect_events()"]
  b9b27172_6269_f5cd_d81e_b606b7c2f1c7 -->|calls| 716d2a5e_dc8e_3cae_e044_b56b06bee655
  style b9b27172_6269_f5cd_d81e_b606b7c2f1c7 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/unit_tests/runnables/test_runnable_events_v2.py lines 2430–2459

async def test_with_explicit_config() -> None:
    """Test astream events with explicit callbacks being passed."""
    infinite_cycle = cycle([AIMessage(content="hello world", id="ai3")])
    model = GenericFakeChatModel(messages=infinite_cycle)

    @tool
    async def say_hello(query: str, callbacks: Callbacks) -> BaseMessage:
        """Use this tool to look up which items are in the given place."""

        @RunnableLambda
        def passthrough_to_trigger_issue(x: str) -> str:
            """Add passthrough to trigger issue."""
            return x

        chain = passthrough_to_trigger_issue | model.with_config(
            {
                "tags": ["hello"],
                "callbacks": callbacks,
            }
        )

        return await chain.ainvoke(query)

    events = await _collect_events(say_hello.astream_events("meow", version="v2"))

    assert [
        event["data"]["chunk"].content
        for event in events
        if event["event"] == "on_chat_model_stream"
    ] == ["hello", " ", "world"]