atrace_as_chain_group() — langchain Function Reference

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

Function python Observability CallbackManager calls 5 called by 1

Entity Profile

Observability→ CallbackManager→ atrace_as_chain_group() — langchain Function Reference

Dependency Diagram

graph TD
  57f655ed_89b3_c486_de7e_73ebe360134a["atrace_as_chain_group()"]
  ef55be46_0333_682d_8311_b4dd35c3e34c["manager.py"]
  57f655ed_89b3_c486_de7e_73ebe360134a -->|defined in| ef55be46_0333_682d_8311_b4dd35c3e34c
  40baea86_dfe8_0c6b_cff8_57b3fde6c2ee["merge()"]
  40baea86_dfe8_0c6b_cff8_57b3fde6c2ee -->|calls| 57f655ed_89b3_c486_de7e_73ebe360134a
  ba6695e3_66d3_2ef4_027e_436cea7aa24c["configure()"]
  57f655ed_89b3_c486_de7e_73ebe360134a -->|calls| ba6695e3_66d3_2ef4_027e_436cea7aa24c
  61bdaf3c_38ed_e800_b567_b934b7b71aaa["on_chain_start()"]
  57f655ed_89b3_c486_de7e_73ebe360134a -->|calls| 61bdaf3c_38ed_e800_b567_b934b7b71aaa
  bb9693c0_b78d_da61_c97e_af8f234e674a["get_child()"]
  57f655ed_89b3_c486_de7e_73ebe360134a -->|calls| bb9693c0_b78d_da61_c97e_af8f234e674a
  3b7be403_b5f2_f25b_2714_6d61955e1332["on_chain_end()"]
  57f655ed_89b3_c486_de7e_73ebe360134a -->|calls| 3b7be403_b5f2_f25b_2714_6d61955e1332
  07a4c44a_3cfb_9b0a_ffde_c1bbfeb4080a["on_chain_error()"]
  57f655ed_89b3_c486_de7e_73ebe360134a -->|calls| 07a4c44a_3cfb_9b0a_ffde_c1bbfeb4080a
  style 57f655ed_89b3_c486_de7e_73ebe360134a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/core/langchain_core/callbacks/manager.py lines 140–215

async def atrace_as_chain_group(
    group_name: str,
    callback_manager: AsyncCallbackManager | None = None,
    *,
    inputs: dict[str, Any] | None = None,
    project_name: str | None = None,
    example_id: str | UUID | None = None,
    run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
) -> AsyncGenerator[AsyncCallbackManagerForChainGroup, None]:
    """Get an async callback manager for a chain group in a context manager.

    Useful for grouping different async calls together as a single run even if they
    aren't composed in a single chain.

    Args:
        group_name: The name of the chain group.
        callback_manager: The async callback manager to use, which manages tracing and
            other callback behavior.
        inputs: The inputs to the chain group.
        project_name: The name of the project.
        example_id: The ID of the example.
        run_id: The ID of the run.
        tags: The inheritable tags to apply to all runs.
        metadata: The metadata to apply to all runs.

    Yields:
        The async callback manager for the chain group.

    !!! note

        Must have `LANGCHAIN_TRACING_V2` env var set to true to see the trace in
        LangSmith.

    Example:
        ```python
        llm_input = "Foo"
        async with atrace_as_chain_group(
            "group_name", inputs={"input": llm_input}
        ) as manager:
            # Use the async callback manager for the chain group
            res = await llm.ainvoke(llm_input, {"callbacks": manager})
            await manager.on_chain_end({"output": res})
        ```
    """
    cb = _get_trace_callbacks(
        project_name, example_id, callback_manager=callback_manager
    )
    cm = AsyncCallbackManager.configure(
        inheritable_callbacks=cb, inheritable_tags=tags, inheritable_metadata=metadata
    )

    run_manager = await cm.on_chain_start(
        {"name": group_name}, inputs or {}, run_id=run_id
    )
    child_cm = run_manager.get_child()
    group_cm = AsyncCallbackManagerForChainGroup(
        child_cm.handlers,
        child_cm.inheritable_handlers,
        child_cm.parent_run_id,
        parent_run_manager=run_manager,
        tags=child_cm.tags,
        inheritable_tags=child_cm.inheritable_tags,
        metadata=child_cm.metadata,
        inheritable_metadata=child_cm.inheritable_metadata,
    )
    try:
        yield group_cm
    except Exception as e:
        if not group_cm.ended:
            await run_manager.on_chain_error(e)
        raise
    else:
        if not group_cm.ended:
            await run_manager.on_chain_end({})