_run_coros() — langchain Function Reference

Architecture documentation for the _run_coros() function in manager.py from the langchain codebase.

Function python CoreAbstractions Serialization called by 1

Entity Profile

CoreAbstractions→ Serialization→ _run_coros() — langchain Function Reference

Dependency Diagram

graph TD
  ab52ff45_f331_7b3a_e93b_9c2558d5c1fb["_run_coros()"]
  35cf5db6_bcb1_b854_6ebb_5e0368e51b58["manager.py"]
  ab52ff45_f331_7b3a_e93b_9c2558d5c1fb -->|defined in| 35cf5db6_bcb1_b854_6ebb_5e0368e51b58
  5f1ef1a4_367e_0b7d_0d39_cf428c8a367e["handle_event()"]
  5f1ef1a4_367e_0b7d_0d39_cf428c8a367e -->|calls| ab52ff45_f331_7b3a_e93b_9c2558d5c1fb
  style ab52ff45_f331_7b3a_e93b_9c2558d5c1fb fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/callbacks/manager.py lines 339–365

def _run_coros(coros: list[Coroutine[Any, Any, Any]]) -> None:
    if hasattr(asyncio, "Runner"):
        # Python 3.11+
        # Run the coroutines in a new event loop, taking care to
        # - install signal handlers
        # - run pending tasks scheduled by `coros`
        # - close asyncgens and executors
        # - close the loop
        with asyncio.Runner() as runner:
            # Run the coroutine, get the result
            for coro in coros:
                try:
                    runner.run(coro)
                except Exception as e:
                    logger.warning("Error in callback coroutine: %s", repr(e))

            # Run pending tasks scheduled by coros until they are all done
            while pending := asyncio.all_tasks(runner.get_loop()):
                runner.run(asyncio.wait(pending))
    else:
        # Before Python 3.11 we need to run each coroutine in a new event loop
        # as the Runner api is not available.
        for coro in coros:
            try:
                asyncio.run(coro)
            except Exception as e:
                logger.warning("Error in callback coroutine: %s", repr(e))