test_async_callbacks_in_sync() — langchain Function Reference

Architecture documentation for the test_async_callbacks_in_sync() function in test_async_callbacks.py from the langchain codebase.

Function python LangChainCore ApiManagement

Entity Profile

LangChainCore→ ApiManagement→ test_async_callbacks_in_sync() — langchain Function Reference

Dependency Diagram

graph TD
  096891e6_bf19_a959_0ee9_261c6d17bd5c["test_async_callbacks_in_sync()"]
  0976f644_8566_afee_cf50_f48330be7127["test_async_callbacks.py"]
  096891e6_bf19_a959_0ee9_261c6d17bd5c -->|defined in| 0976f644_8566_afee_cf50_f48330be7127
  style 096891e6_bf19_a959_0ee9_261c6d17bd5c fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/tests/benchmarks/test_async_callbacks.py lines 48–56

async def test_async_callbacks_in_sync(benchmark: BenchmarkFixture) -> None:
    infinite_cycle = cycle([AIMessage(content=" ".join(["hello", "goodbye"] * 5))])
    model = GenericFakeChatModel(messages=infinite_cycle)

    @benchmark  # type: ignore[untyped-decorator]
    def sync_callbacks() -> None:
        for _ in range(5):
            for _ in model.stream("meow", {"callbacks": [MyCustomAsyncHandler()]}):
                pass