flowchart TB
    subgraph UserLayer["用户层"]
        APP[应用代码<br/>FastAPI/Django/CLI]
        LCEL[LCEL表达式<br/>prompt | model | parser]
    end

    subgraph InterfaceLayer["接口层"]
        RUNNABLE[Runnable抽象基类<br/>定义统一协议]
        API_INVOKE[invoke/ainvoke]
        API_BATCH[batch/abatch]
        API_STREAM[stream/astream]
    end

    subgraph CompositionLayer["组合层"]
        SEQ[RunnableSequence<br/>顺序执行]
        PAR[RunnableParallel<br/>并行执行]
        BRANCH[RunnableBranch<br/>条件分支]
        LAMBDA[RunnableLambda<br/>函数包装]
    end

    subgraph DecoratorLayer["装饰层"]
        RETRY[RunnableRetry<br/>重试机制]
        FALLBACK[RunnableWithFallbacks<br/>故障转移]
        BINDING[RunnableBinding<br/>参数绑定]
        HISTORY[RunnableWithMessageHistory<br/>历史注入]
    end

    subgraph ExecutionLayer["执行层"]
        CALLBACK[CallbackManager<br/>回调管理]
        CONFIG[RunnableConfig<br/>配置传递]
        EXECUTOR[ThreadPoolExecutor<br/>并发控制]
        ASYNC[AsyncIO EventLoop<br/>异步调度]
    end

    subgraph ObservabilityLayer["可观测层"]
        TRACER[LangSmithTracer<br/>追踪记录]
        METRICS[Metrics<br/>性能指标]
        LOGS[Logs<br/>日志输出]
    end

    APP --> LCEL
    LCEL --> RUNNABLE
    RUNNABLE --> API_INVOKE
    RUNNABLE --> API_BATCH
    RUNNABLE --> API_STREAM

    API_INVOKE --> SEQ
    API_INVOKE --> PAR
    API_INVOKE --> BRANCH
    API_INVOKE --> LAMBDA

    SEQ --> RETRY
    PAR --> RETRY
    BRANCH --> FALLBACK
    LAMBDA --> BINDING

    RETRY --> CALLBACK
    FALLBACK --> CALLBACK
    BINDING --> CONFIG
    HISTORY --> CONFIG

    CALLBACK --> EXECUTOR
    CONFIG --> EXECUTOR
    CALLBACK --> ASYNC
    CONFIG --> ASYNC

    EXECUTOR --> TRACER
    ASYNC --> TRACER
    TRACER --> METRICS
    TRACER --> LOGS

    style UserLayer fill:#e3f2fd
    style InterfaceLayer fill:#fff3e0
    style CompositionLayer fill:#e8f5e9
    style DecoratorLayer fill:#fce4ec
    style ExecutionLayer fill:#f3e5f5
    style ObservabilityLayer fill:#e0f2f1

flowchart TB
    subgraph Interface["核心接口层"]
        RUNNABLE[Runnable抽象基类<br/>定义统一协议]
    end

    subgraph Serializable["序列化层"]
        RUNNABLE_SER[RunnableSerializable<br/>可序列化Runnable]
    end

    subgraph Composition["组合原语"]
        SEQ[RunnableSequence<br/>顺序执行 A|B|C]
        PAR[RunnableParallel<br/>并行执行 {a:A,b:B}]
        BRANCH[RunnableBranch<br/>条件分支]
        LAMBDA[RunnableLambda<br/>包装函数]
        GEN[RunnableGenerator<br/>生成器包装]
    end

    subgraph Binding["绑定与装饰"]
        BINDING[RunnableBinding<br/>绑定参数]
        FALLBACK[RunnableWithFallbacks<br/>故障转移]
        RETRY[RunnableRetry<br/>重试机制]
        HISTORY[RunnableWithMessageHistory<br/>历史记录]
    end

    subgraph Passthrough["数据操作"]
        PASS[RunnablePassthrough<br/>透传数据]
        ASSIGN[RunnableAssign<br/>添加字段]
        PICK[RunnablePick<br/>选择字段]
    end

    subgraph Config["配置系统"]
        RCONFIG[RunnableConfig<br/>配置字典]
        CONFIG_FIELD[ConfigurableField<br/>可配置字段]
        CONFIG_ALT[ConfigurableAlternatives<br/>可配置备选]
    end

    subgraph Utils["工具类"]
        EACH[RunnableEach<br/>映射到列表]
        ROUTER[RouterRunnable<br/>路由器]
        ADDABLE[AddableDict<br/>可加字典]
    end

    RUNNABLE -.继承.-> RUNNABLE_SER
    RUNNABLE_SER -.继承.-> SEQ
    RUNNABLE_SER -.继承.-> PAR
    RUNNABLE_SER -.继承.-> BRANCH
    RUNNABLE -.继承.-> LAMBDA
    RUNNABLE -.继承.-> GEN

    RUNNABLE_SER -.继承.-> BINDING
    BINDING -.继承.-> FALLBACK
    BINDING -.继承.-> RETRY
    BINDING -.继承.-> HISTORY

    RUNNABLE_SER -.继承.-> PASS
    RUNNABLE_SER -.继承.-> ASSIGN
    RUNNABLE_SER -.继承.-> PICK

    RUNNABLE_SER -.继承.-> EACH
    RUNNABLE_SER -.继承.-> ROUTER

    RUNNABLE --> RCONFIG
    BINDING --> CONFIG_FIELD
    BINDING --> CONFIG_ALT

    style RUNNABLE fill:#e1f5ff
    style RUNNABLE_SER fill:#fff4e1
    style SEQ fill:#e8f5e9
    style PAR fill:#e8f5e9
    style FALLBACK fill:#ffe8e8
    style RETRY fill:#ffe8e8

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant LCEL as LCEL表达式
    participant Seq as RunnableSequence
    participant Config as RunnableConfig
    participant CB as CallbackManager
    participant Comp as 组件Runnable
    participant Tracer as LangSmithTracer
    participant Executor as ThreadPoolExecutor

    User->>LCEL: chain = prompt | model | parser
    LCEL->>Seq: 创建RunnableSequence
    Note over Seq: 组合三个Runnable

    User->>Seq: invoke(input, config)
    Seq->>Config: ensure_config(config)
    Config-->>Seq: 规范化后的config

    Seq->>CB: get_callback_manager_for_config(config)
    CB->>Tracer: 创建追踪器实例
    CB-->>Seq: callback_manager

    Seq->>CB: on_chain_start(self, input)
    CB->>Tracer: 记录开始事件
    CB-->>Seq: run_manager

    loop 遍历每个步骤
        Seq->>Comp: invoke(input, patched_config)
        Comp->>CB: on_llm_start()/on_tool_start()
        Comp->>Comp: 执行核心逻辑
        Comp->>CB: on_llm_end()/on_tool_end()
        Comp-->>Seq: output
        Note over Seq: output成为下一步input
    end

    Seq->>CB: on_chain_end(final_output)
    CB->>Tracer: 记录完成事件
    Seq-->>User: final_output

    Note over User,Executor: 批处理路径
    User->>Seq: batch([input1, input2, ...])
    Seq->>Config: get_config_list(config, len(inputs))
    Seq->>Executor: 创建线程池

    par 并行执行
        Seq->>Comp: invoke(input1, config1)
        Seq->>Comp: invoke(input2, config2)
    end

    Seq-->>User: [output1, output2, ...]

# 用户代码
chain = prompt_template | chat_model | output_parser
result = chain.invoke({"topic": "AI"})

graph TD
    A[chain.invoke<br/>RunnableSequence] --> B[ensure_config<br/>规范化配置]
    B --> C[get_callback_manager<br/>获取回调管理器]
    C --> D[on_chain_start<br/>开始追踪]
    D --> E1[步骤1: prompt_template.invoke]
    E1 --> E1A[format_messages<br/>格式化模板]
    E1A --> E1B[返回 List<BaseMessage>]

    E1B --> E2[步骤2: chat_model.invoke]
    E2 --> E2A[_generate_with_cache<br/>检查缓存]
    E2A --> E2B{缓存命中?}
    E2B -->|是| E2C[返回缓存结果]
    E2B -->|否| E2D[_generate<br/>调用LLM API]
    E2D --> E2E[on_llm_start/on_llm_end<br/>触发回调]
    E2E --> E2F[返回 AIMessage]

    E2C --> E3[步骤3: output_parser.invoke]
    E2F --> E3
    E3 --> E3A[parse<br/>解析输出]
    E3A --> E3B[返回结构化数据]

    E3B --> F[on_chain_end<br/>完成追踪]
    F --> G[返回最终结果]

    style A fill:#e3f2fd
    style E1 fill:#fff3e0
    style E2 fill:#e8f5e9
    style E3 fill:#fce4ec
    style G fill:#e0f2f1

def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
    # 阶段1: 初始化配置和回调
    config = ensure_config(config)  # 填充默认值
    callback_manager = get_callback_manager_for_config(config)

    # 阶段2: 开始链式调用追踪
    run_manager = callback_manager.on_chain_start(
        dumpd(self),  # 序列化自身用于追踪
        input,
        name=config.get("run_name") or self.get_name(),
        run_id=config.get("run_id"),
    )

    # 阶段3: 逐步执行
    try:
        for i, step in enumerate(self.steps):
            # 为每个步骤创建子config
            step_config = patch_config(
                config,
                callbacks=run_manager.get_child(),  # 父子关系
                run_name=f"{self.get_name()}.step[{i}]"
            )
            # 前一步输出作为下一步输入
            input = step.invoke(input, step_config)
    except BaseException as e:
        # 阶段4a: 异常处理
        run_manager.on_chain_error(e)
        raise
    else:
        # 阶段4b: 正常完成
        run_manager.on_chain_end(input)
        return input

def ensure_config(config: RunnableConfig | None = None) -> RunnableConfig:
    """确保配置对象存在且包含所有必需字段"""
    empty = RunnableConfig(
        tags=[],
        metadata={},
        callbacks=None,
        recursion_limit=25,  # 默认递归限制
        configurable={},
    )

    if config is None:
        return empty

    # 合并用户配置和默认值
    return merge_configs(empty, config)

def patch_config(
    config: RunnableConfig,
    *,
    callbacks: Callbacks = None,
    recursion_limit: int | None = None,
    max_concurrency: int | None = None,
    run_name: str | None = None,
) -> RunnableConfig:
    """在现有配置基础上修补字段,用于子调用"""
    patched = config.copy()

    if callbacks is not None:
        patched["callbacks"] = callbacks
    if recursion_limit is not None:
        patched["recursion_limit"] = recursion_limit
    if max_concurrency is not None:
        patched["max_concurrency"] = max_concurrency
    if run_name is not None:
        patched["run_name"] = run_name

    return patched

def on_chain_start(
    self,
    serialized: dict[str, Any],
    inputs: dict[str, Any],
    run_id: UUID | None = None,
    **kwargs: Any,
) -> CallbackManagerForChainRun:
    """链开始时触发回调"""
    run_id = run_id or uuid.uuid4()

    # 遍历所有注册的回调处理器
    for handler in self.handlers:
        # 异步执行,不阻塞主流程
        handler.on_chain_start(
            serialized,
            inputs,
            run_id=run_id,
            parent_run_id=self.parent_run_id,
            tags=self.tags,
            metadata=self.metadata,
            **kwargs
        )

    # 返回子运行管理器
    return CallbackManagerForChainRun(
        run_id=run_id,
        handlers=self.handlers,
        inheritable_handlers=self.inheritable_handlers,
        parent_run_id=self.parent_run_id,
        tags=self.tags,
        inheritable_tags=self.inheritable_tags,
        metadata=self.metadata,
        inheritable_metadata=self.inheritable_metadata,
    )

# 用户代码
parallel_chain = RunnableParallel({
    "joke": ChatPromptTemplate.from_template("tell a joke about {topic}") | model,
    "poem": ChatPromptTemplate.from_template("write a poem about {topic}") | model,
    "fact": ChatPromptTemplate.from_template("tell a fact about {topic}") | model,
})
result = parallel_chain.invoke({"topic": "cats"})
# 返回: {"joke": "...", "poem": "...", "fact": "..."}

graph TD
    A[parallel_chain.invoke<br/>RunnableParallel] --> B[ensure_config]
    B --> C[get_callback_manager]
    C --> D[on_chain_start]

    D --> E[创建ThreadPoolExecutor]
    E --> F{并发执行所有分支}

    F --> G1[分支1: joke_chain.invoke]
    F --> G2[分支2: poem_chain.invoke]
    F --> G3[分支3: fact_chain.invoke]

    G1 --> G1A[RunnableSequence执行]
    G2 --> G2A[RunnableSequence执行]
    G3 --> G3A[RunnableSequence执行]

    G1A --> H1[返回joke结果]
    G2A --> H2[返回poem结果]
    G3A --> H3[返回fact结果]

    H1 --> I[等待所有Future完成]
    H2 --> I
    H3 --> I

    I --> J[合并结果为字典]
    J --> K[on_chain_end]
    K --> L[返回最终字典]

    style A fill:#e3f2fd
    style E fill:#fff3e0
    style G1 fill:#e8f5e9
    style G2 fill:#e8f5e9
    style G3 fill:#e8f5e9
    style L fill:#e0f2f1

def invoke(self, input: Input, config: RunnableConfig | None = None) -> dict[str, Any]:
    # 阶段1: 初始化
    config = ensure_config(config)
    callback_manager = get_callback_manager_for_config(config)
    run_manager = callback_manager.on_chain_start(dumpd(self), input)

    # 阶段2: 准备并发执行
    max_concurrency = config.get("max_concurrency")

    try:
        # 阶段3: 并发执行所有分支
        with ThreadPoolExecutor(max_workers=max_concurrency) as executor:
            # 为每个分支提交任务
            futures = {
                key: executor.submit(
                    step.invoke,
                    input,  # 所有分支共享相同输入
                    patch_config(config, callbacks=run_manager.get_child())
                )
                for key, step in self.steps__.items()
            }

            # 阶段4: 收集结果
            output = {}
            for key, future in futures.items():
                try:
                    output[key] = future.result()  # 阻塞等待结果
                except Exception as e:
                    # 单个分支失败不影响其他分支(取决于配置)
                    run_manager.on_chain_error(e)
                    raise
    except BaseException as e:
        run_manager.on_chain_error(e)
        raise
    else:
        run_manager.on_chain_end(output)
        return output

# ThreadPoolExecutor内部实现
class ThreadPoolExecutor:
    def submit(self, fn, /, *args, **kwargs):
        """提交任务到线程池"""
        # 1. 创建Future对象
        future = Future()

        # 2. 包装任务
        work_item = _WorkItem(future, fn, args, kwargs)

        # 3. 放入工作队列
        self._work_queue.put(work_item)

        # 4. 唤醒工作线程
        self._adjust_thread_count()

        return future

    def _worker(self):
        """工作线程循环"""
        while True:
            work_item = self._work_queue.get(block=True)
            if work_item is not None:
                try:
                    result = work_item.fn(*work_item.args, **work_item.kwargs)
                    work_item.future.set_result(result)
                except Exception as exc:
                    work_item.future.set_exception(exc)

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Par as RunnableParallel
    participant Executor as ThreadPoolExecutor
    participant Thread1 as 工作线程1
    participant Thread2 as 工作线程2
    participant Thread3 as 工作线程3
    participant Branch1 as joke_chain
    participant Branch2 as poem_chain
    participant Branch3 as fact_chain

    User->>Par: invoke({"topic": "cats"})
    Par->>Par: ensure_config & get_callback_manager
    Par->>Par: on_chain_start

    Par->>Executor: 创建线程池(max_workers=3)
    Executor->>Thread1: 启动工作线程
    Executor->>Thread2: 启动工作线程
    Executor->>Thread3: 启动工作线程

    Par->>Executor: submit(joke_chain.invoke, input)
    Par->>Executor: submit(poem_chain.invoke, input)
    Par->>Executor: submit(fact_chain.invoke, input)

    Note over Executor: 任务分发到工作队列

    par 并行执行三个分支
        Thread1->>Branch1: invoke(input, config)
        Branch1->>Branch1: 执行prompt | model链
        Branch1-->>Thread1: "Why did the cat..."

        Thread2->>Branch2: invoke(input, config)
        Branch2->>Branch2: 执行prompt | model链
        Branch2-->>Thread2: "Whiskers soft..."

        Thread3->>Branch3: invoke(input, config)
        Branch3->>Branch3: 执行prompt | model链
        Branch3-->>Thread3: "Cats can rotate..."
    end

    Thread1-->>Par: future1.result()
    Thread2-->>Par: future2.result()
    Thread3-->>Par: future3.result()

    Par->>Par: 合并结果为字典
    Par->>Par: on_chain_end(output)
    Par-->>User: {"joke": "...", "poem": "...", "fact": "..."}

# 异步应用代码
import asyncio
from fastapi import FastAPI

app = FastAPI()
chain = prompt | model | parser

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    # 使用ainvoke进行非阻塞调用
    result = await chain.ainvoke(
        {"messages": request.messages},
        config={"metadata": {"user_id": request.user_id}}
    )
    return {"response": result}

graph TD
    A[chain.ainvoke<br/>RunnableSequence] --> B[ensure_config]
    B --> C[get_async_callback_manager]
    C --> D[await on_chain_start<br/>异步回调]

    D --> E1[await step1.ainvoke<br/>异步执行]
    E1 --> E1A[await format_messages_async]
    E1A --> E1B[返回messages]

    E1B --> E2[await step2.ainvoke<br/>ChatModel]
    E2 --> E2A[await _agenerate<br/>异步HTTP请求]
    E2A --> E2B{使用httpx/aiohttp}
    E2B --> E2C[await response]
    E2C --> E2D[返回AIMessage]

    E2D --> E3[await step3.ainvoke<br/>Parser]
    E3 --> E3A[await parse_async]
    E3A --> E3B[返回结构化数据]

    E3B --> F[await on_chain_end]
    F --> G[返回最终结果]

    style A fill:#e3f2fd
    style E2A fill:#fff3e0
    style E2B fill:#e8f5e9
    style G fill:#e0f2f1

async def ainvoke(
    self,
    input: Input,
    config: RunnableConfig | None = None
) -> Output:
    # 阶段1: 异步初始化
    config = ensure_config(config)
    callback_manager = get_async_callback_manager_for_config(config)

    # 阶段2: 异步开始追踪
    run_manager = await callback_manager.on_chain_start(
        dumpd(self),
        input,
        name=config.get("run_name") or self.get_name(),
    )

    # 阶段3: 异步逐步执行
    try:
        for i, step in enumerate(self.steps):
            step_config = patch_config(
                config,
                callbacks=run_manager.get_child()
            )
            # 使用await等待异步执行
            input = await step.ainvoke(input, step_config)
    except BaseException as e:
        await run_manager.on_chain_error(e)
        raise
    else:
        await run_manager.on_chain_end(input)
        return input

class AsyncCallbackManager:
    async def on_chain_start(
        self,
        serialized: dict[str, Any],
        inputs: dict[str, Any],
        **kwargs: Any,
    ) -> AsyncCallbackManagerForChainRun:
        """异步触发回调"""
        run_id = kwargs.get("run_id") or uuid.uuid4()

        # 并发触发所有回调处理器
        await asyncio.gather(
            *[
                handler.on_chain_start(
                    serialized,
                    inputs,
                    run_id=run_id,
                    parent_run_id=self.parent_run_id,
                    tags=self.tags,
                    metadata=self.metadata,
                    **kwargs
                )
                for handler in self.handlers
                if hasattr(handler, 'on_chain_start')
            ],
            return_exceptions=True  # 单个回调失败不影响其他
        )

        return AsyncCallbackManagerForChainRun(
            run_id=run_id,
            handlers=self.handlers,
            # ... 其他字段
        )

class ChatOpenAI(BaseChatModel):
    async def ainvoke(
        self,
        input: list[BaseMessage],
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> AIMessage:
        # 使用异步HTTP客户端
        async with httpx.AsyncClient() as client:
            response = await client.post(
                self.openai_api_base,
                json=self._prepare_request(input),
                headers=self._get_headers(),
                timeout=self.request_timeout,
            )
            response.raise_for_status()
            return self._parse_response(response.json())

sequenceDiagram
    autonumber
    participant User as 异步应用
    participant FastAPI
    participant EventLoop as asyncio事件循环
    participant Seq as RunnableSequence
    participant LLM as ChatOpenAI
    participant HTTP as httpx.AsyncClient
    participant API as OpenAI API

    User->>FastAPI: POST /chat (HTTP请求)
    FastAPI->>EventLoop: 创建Task(chat_endpoint)
    EventLoop->>Seq: await chain.ainvoke(input)

    Seq->>Seq: ensure_config
    Seq->>Seq: await on_chain_start

    Seq->>LLM: await llm.ainvoke(messages)
    LLM->>HTTP: await client.post(url, json=...)

    Note over EventLoop: 释放事件循环<br/>可处理其他请求

    HTTP->>API: HTTP POST请求

    Note over API: OpenAI处理请求<br/>生成响应(5-10秒)

    API-->>HTTP: HTTP 200 OK (响应)
    HTTP-->>LLM: response.json()
    LLM->>LLM: parse_response
    LLM-->>Seq: AIMessage

    Seq->>Seq: await on_chain_end
    Seq-->>EventLoop: 完成Task
    EventLoop-->>FastAPI: 返回响应
    FastAPI-->>User: HTTP 200 (JSON响应)

    Note over User,API: 在等待API响应期间<br/>事件循环可处理其他请求<br/>实现高并发

async def abatch(
    self,
    inputs: list[Input],
    config: RunnableConfig | list[RunnableConfig] | None = None,
    **kwargs: Any,
) -> list[Output]:
    configs = get_config_list(config, len(inputs))

    # 使用asyncio.gather并发执行所有ainvoke
    results = await asyncio.gather(
        *[
            self.ainvoke(input, config)
            for input, config in zip(inputs, configs)
        ],
        return_exceptions=kwargs.get("return_exceptions", False)
    )

    return results

# 构建容错链
primary_model = ChatOpenAI(model="gpt-4o")
fallback_model = ChatOpenAI(model="gpt-4o-mini")

# 添加重试和备用
robust_model = primary_model.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
).with_fallbacks([fallback_model])

chain = prompt | robust_model | parser
result = chain.invoke({"topic": "AI"})

graph TD
    A[chain.invoke] --> B[RunnableSequence.invoke]
    B --> C[prompt.invoke]
    C --> D[RunnableWithFallbacks.invoke]

    D --> E[主路径: RunnableRetry.invoke]
    E --> F{尝试1: primary_model.invoke}
    F -->|失败| G{等待2^1秒+jitter}
    G --> H{尝试2: primary_model.invoke}
    H -->|失败| I{等待2^2秒+jitter}
    I --> J{尝试3: primary_model.invoke}
    J -->|仍失败| K[抛出异常]

    K --> L[RunnableWithFallbacks捕获异常]
    L --> M[切换到fallback: fallback_model.invoke]
    M -->|成功| N[返回AIMessage]

    J -->|成功| N
    F -->|成功| N
    H -->|成功| N

    N --> O[parser.invoke]
    O --> P[返回最终结果]

    style D fill:#ffe8e8
    style E fill:#fff3e0
    style M fill:#e8f5e9
    style P fill:#e0f2f1

class RunnableRetry(RunnableBinding[Input, Output]):
    max_attempt_number: int = 3
    wait_exponential_jitter: bool = True
    retry_exception_types: tuple[type[BaseException], ...] = (Exception,)

    def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
        from tenacity import (
            retry,
            stop_after_attempt,
            wait_exponential_jitter,
            retry_if_exception_type,
        )

        # 使用tenacity库实现重试装饰器
        @retry(
            stop=stop_after_attempt(self.max_attempt_number),
            wait=wait_exponential_jitter() if self.wait_exponential_jitter
                 else wait_exponential(),
            retry=retry_if_exception_type(self.retry_exception_types),
            before_sleep=self._log_retry,
            reraise=True,
        )
        def _invoke_with_retry():
            return self.bound.invoke(input, config)

        return _invoke_with_retry()

    def _log_retry(self, retry_state):
        """重试前记录日志"""
        print(f"Attempt {retry_state.attempt_number} failed, retrying...")

class RunnableWithFallbacks(RunnableSerializable[Input, Output]):
    def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
        config = ensure_config(config)
        first_error = None

        # 尝试主Runnable
        try:
            return self.runnable.invoke(input, config)
        except self.exceptions_to_handle as e:
            first_error = e

        # 依次尝试每个fallback
        for fallback in self.fallbacks:
            try:
                return fallback.invoke(input, config)
            except self.exceptions_to_handle as e:
                # 记录但继续尝试下一个
                continue

        # 所有选项都失败,抛出最初的异常
        raise first_error

sequenceDiagram
    autonumber
    participant User
    participant Fallbacks as RunnableWithFallbacks
    participant Retry as RunnableRetry
    participant Primary as primary_model
    participant Fallback as fallback_model
    participant API1 as OpenAI API(gpt-4o)
    participant API2 as OpenAI API(gpt-4o-mini)

    User->>Fallbacks: invoke(input)
    Fallbacks->>Retry: invoke(input)

    Note over Retry: 第1次尝试
    Retry->>Primary: invoke(input)
    Primary->>API1: HTTP POST
    API1-->>Primary: 503 Service Unavailable
    Primary-->>Retry: APIError

    Note over Retry: 等待2^1秒(2s) + 随机抖动

    Note over Retry: 第2次尝试
    Retry->>Primary: invoke(input)
    Primary->>API1: HTTP POST
    API1-->>Primary: 429 Rate Limit
    Primary-->>Retry: RateLimitError

    Note over Retry: 等待2^2秒(4s) + 随机抖动

    Note over Retry: 第3次尝试
    Retry->>Primary: invoke(input)
    Primary->>API1: HTTP POST
    API1-->>Primary: Timeout
    Primary-->>Retry: TimeoutError

    Retry-->>Fallbacks: 抛出TimeoutError

    Note over Fallbacks: 主路径失败<br/>切换到fallback

    Fallbacks->>Fallback: invoke(input)
    Fallback->>API2: HTTP POST
    API2-->>Fallback: 200 OK
    Fallback-->>Fallbacks: AIMessage

    Fallbacks-->>User: 返回结果(来自fallback)

class MyRunnable(Runnable[InputType, OutputType]):
    def invoke(self, input: InputType, config: RunnableConfig) -> OutputType:
        # 实现核心逻辑
        pass

runnable.configurable_fields(
    temperature=ConfigurableField(id="temperature", name="Temperature")
).configurable_alternatives(
    ConfigurableField(id="model"),
    default_key="gpt-4o",
    gpt35=ChatOpenAI(model="gpt-3.5-turbo")
)

classDiagram
    class Runnable~Input, Output~ {
        <<abstract>>
        +name: str|None
        +invoke(input: Input, config: RunnableConfig) Output
        +ainvoke(input: Input, config: RunnableConfig) Awaitable[Output]
        +batch(inputs: list[Input], config: RunnableConfig) list[Output]
        +abatch(inputs: list[Input], config: RunnableConfig) Awaitable~list[Output]~
        +stream(input: Input, config: RunnableConfig) Iterator[Output]
        +astream(input: Input, config: RunnableConfig) AsyncIterator[Output]
        +InputType: type[Input]
        +OutputType: type[Output]
        +input_schema: type[BaseModel]
        +output_schema: type[BaseModel]
        +with_retry(...) RunnableRetry
        +with_fallbacks(...) RunnableWithFallbacks
        +with_config(...) RunnableBinding
        +pipe(...) RunnableSequence
        +__or__(other) RunnableSequence
    }

    class RunnableConfig {
        <<TypedDict>>
        +tags: list[str]
        +metadata: dict
        +callbacks: Callbacks
        +run_name: str
        +max_concurrency: int|None
        +recursion_limit: int
        +configurable: dict
        +run_id: UUID|None
    }

    class RunnableSerializable~Input, Output~ {
        +lc_serializable: bool
        +get_lc_namespace() list[str]
        +to_json() dict
        +save(path: str) None
    }

    class RunnableSequence~Input, Output~ {
        +first: Runnable[Input, Any]
        +middle: list[Runnable]
        +last: Runnable[Any, Output]
        +steps: list[Runnable]
    }

    class RunnableParallel~Input~ {
        +steps: dict[str, Runnable[Input, Any]]
    }

    class RunnableBranch~Input, Output~ {
        +branches: list[tuple[Callable, Runnable]]
        +default: Runnable[Input, Output]
    }

    class RunnableLambda~Input, Output~ {
        +func: Callable
        +afunc: Callable|None
    }

    class RunnableBinding~Input, Output~ {
        +bound: Runnable[Input, Output]
        +kwargs: dict
        +config: RunnableConfig
        +config_factories: list[Callable]
    }

    class RunnableWithFallbacks~Input, Output~ {
        +runnable: Runnable[Input, Output]
        +fallbacks: list[Runnable[Input, Output]]
    }

    Runnable <|-- RunnableSerializable
    RunnableSerializable <|-- RunnableSequence
    RunnableSerializable <|-- RunnableParallel
    RunnableSerializable <|-- RunnableBranch
    Runnable <|-- RunnableLambda
    RunnableSerializable <|-- RunnableBinding
    RunnableSerializable <|-- RunnableWithFallbacks

    Runnable --> RunnableConfig : uses

# invoke方法参数
input: Input  # 泛型输入,类型由子类指定
config: RunnableConfig | None  # 可选配置

# invoke方法返回值
output: Output  # 泛型输出,类型由子类指定

# Runnable基类的invoke方法(抽象)
@abstractmethod
def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
    """将单个输入转换为输出。

    Args:
        input: 输入数据
        config: 可选的运行时配置

    Returns:
        转换后的输出数据
    """
    ...

# RunnableSequence的invoke实现
class RunnableSequence(RunnableSerializable[Input, Output]):
    def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
        # 1) 确保config存在并设置上下文
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)

        # 2) 开始链式调用追踪
        run_manager = callback_manager.on_chain_start(
            dumpd(self),
            input,
            name=config.get("run_name") or self.get_name()
        )

        # 3) 依次调用序列中的每个步骤
        try:
            for step in self.steps:
                input = step.invoke(
                    input,
                    patch_config(config, callbacks=run_manager.get_child())
                )
        except BaseException as e:
            run_manager.on_chain_error(e)
            raise
        else:
            run_manager.on_chain_end(input)
            return input

# 用户代码调用
chain = prompt | model | parser
result = chain.invoke({"topic": "AI"})

# 调用栈
# 1. RunnableSequence.invoke (chain)
#    ├─ 2. PromptTemplate.invoke (prompt)
#    │    └─ 格式化模板,返回messages
#    ├─ 3. ChatModel.invoke (model)
#    │    ├─ _generate_with_cache (缓存逻辑)
#    │    ├─ _generate (实际API调用)
#    │    └─ 返回AIMessage
#    └─ 4. OutputParser.invoke (parser)
#         └─ 解析message.content,返回结构化输出

# 上层适配(Web框架中)
from fastapi import FastAPI
app = FastAPI()

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    # 调用链的invoke
    result = chain.invoke(
        {"messages": request.messages},
        config={
            "metadata": {"user_id": request.user_id},
            "tags": ["api", "prod"]
        }
    )
    return {"response": result}

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant Seq as RunnableSequence
    participant CB as CallbackManager
    participant Step1 as Prompt
    participant Step2 as LLM
    participant Step3 as Parser

    User->>Seq: invoke(input, config)
    Seq->>Seq: ensure_config(config)
    Seq->>CB: get_callback_manager_for_config
    CB-->>Seq: callback_manager

    Seq->>CB: on_chain_start(self, input)
    CB-->>Seq: run_manager

    Seq->>Step1: invoke(input, config)
    Step1->>Step1: 格式化模板
    Step1-->>Seq: messages

    Seq->>Step2: invoke(messages, config)
    Step2->>CB: on_llm_start(messages)
    Step2->>Step2: 调用LLM API
    Step2->>CB: on_llm_end(response)
    Step2-->>Seq: AIMessage

    Seq->>Step3: invoke(AIMessage, config)
    Step3->>Step3: 解析输出
    Step3-->>Seq: parsed_output

    Seq->>CB: on_chain_end(parsed_output)
    Seq-->>User: final_output

inputs: list[Input]  # 输入列表
config: RunnableConfig | list[RunnableConfig] | None  # 配置,可为每个输入单独配置
return_exceptions: bool = False  # 是否返回异常对象而非抛出
max_concurrency: int | None  # 通过config.max_concurrency或kwargs传递

outputs: list[Output | Exception]  # 输出列表,顺序与输入对应

class Runnable(ABC, Generic[Input, Output]):
    def batch(
        self,
        inputs: list[Input],
        config: RunnableConfig | list[RunnableConfig] | None = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Any,
    ) -> list[Output]:
        # 1) 规范化config为列表
        configs = get_config_list(config, len(inputs))

        # 2) 提取max_concurrency
        max_concurrency = (
            kwargs.get("max_concurrency")
            or configs[0].get("max_concurrency")
            if configs
            else None
        )

        # 3) 使用线程池并行执行
        with ThreadPoolExecutor(max_workers=max_concurrency) as executor:
            futures = [
                executor.submit(self.invoke, input, config)
                for input, config in zip(inputs, configs)
            ]

            # 4) 收集结果
            results = []
            for future in futures:
                try:
                    results.append(future.result())
                except Exception as e:
                    if return_exceptions:
                        results.append(e)
                    else:
                        raise

            return results

# 批量处理多个请求
topics = ["AI", "区块链", "量子计算"]
results = chain.batch([{"topic": t} for t in topics])

# 限制并发
results = chain.batch(
    [{"topic": t} for t in topics],
    config={"max_concurrency": 2}  # 最多2个并发
)

# 容错批处理
results = chain.batch(
    inputs,
    return_exceptions=True  # 失败项返回异常对象
)
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"输入{i}失败: {result}")
    else:
        print(f"输入{i}成功: {result}")

sequenceDiagram
    autonumber
    participant User
    participant Runnable
    participant ThreadPool
    participant Worker1
    participant Worker2
    participant WorkerN

    User->>Runnable: batch([input1, input2, ..., inputN], config)
    Runnable->>Runnable: get_config_list(config, N)
    Runnable->>ThreadPool: 创建线程池(max_workers=max_concurrency)

    par 并行执行
        Runnable->>Worker1: submit(invoke, input1)
        Runnable->>Worker2: submit(invoke, input2)
        Runnable->>WorkerN: submit(invoke, inputN)
    end

    Worker1->>Worker1: invoke(input1)
    Worker2->>Worker2: invoke(input2)
    WorkerN->>WorkerN: invoke(inputN)

    Worker1-->>Runnable: output1
    Worker2-->>Runnable: output2
    WorkerN-->>Runnable: outputN

    Runnable->>Runnable: 按输入顺序收集结果
    Runnable-->>User: [output1, output2, ..., outputN]

input: Input
config: RunnableConfig | None

# 返回迭代器,yield输出块
Iterator[Output]

class Runnable(ABC, Generic[Input, Output]):
    def stream(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Iterator[Output]:
        # 默认实现:调用invoke并yield整个结果
        # 子类可重写以实现真正的流式输出
        yield self.invoke(input, config, **kwargs)

# RunnableSequence的流式实现
class RunnableSequence(RunnableSerializable[Input, Output]):
    def stream(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Iterator[Output]:
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)
        run_manager = callback_manager.on_chain_start(
            dumpd(self),
            input,
        )

        # 1) 执行前N-1个步骤(非流式)
        for step in self.steps[:-1]:
            input = step.invoke(input, patch_config(config, callbacks=run_manager.get_child()))

        # 2) 最后一个步骤流式执行
        final_pipeline = self.steps[-1].stream(
            input,
            patch_config(config, callbacks=run_manager.get_child()),
        )

        # 3) yield最后一步的每个chunk
        for chunk in final_pipeline:
            yield chunk

        run_manager.on_chain_end(chunk)

# 流式输出示例
for chunk in chain.stream({"topic": "量子计算"}):
    print(chunk, end="", flush=True)

# Web应用中的流式响应
from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    def generate():
        for chunk in chain.stream({"messages": request.messages}):
            yield f"data: {json.dumps(chunk)}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

sequenceDiagram
    autonumber
    participant User
    participant Seq as RunnableSequence
    participant Step1 as Prompt
    participant Step2 as LLM

    User->>Seq: stream(input)
    Seq->>Step1: invoke(input)
    Step1-->>Seq: formatted_prompt

    Seq->>Step2: stream(formatted_prompt)
    Step2->>Step2: 开始LLM流式生成

    loop 每个token
        Step2-->>Seq: chunk1
        Seq-->>User: chunk1
        Note over User: 立即显示
        Step2-->>Seq: chunk2
        Seq-->>User: chunk2
        Step2-->>Seq: chunk3
        Seq-->>User: chunk3
    end

    Step2-->>Seq: 流结束
    Seq-->>User: 迭代器关闭

# 返回Awaitable
Awaitable[Output]

class Runnable(ABC, Generic[Input, Output]):
    async def ainvoke(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Output:
        # 默认实现:在executor中运行同步invoke
        return await run_in_executor(
            config,
            self.invoke,
            input,
            config,
            **kwargs
        )

# RunnableSequence的异步实现
class RunnableSequence(RunnableSerializable[Input, Output]):
    async def ainvoke(
        self,
        input: Input,
        config: RunnableConfig | None = None,
        **kwargs: Any | None,
    ) -> Output:
        config = ensure_config(config)
        callback_manager = get_async_callback_manager_for_config(config)
        run_manager = await callback_manager.on_chain_start(
            dumpd(self),
            input,
        )

        # 依次异步调用每个步骤
        try:
            for step in self.steps:
                input = await step.ainvoke(
                    input,
                    patch_config(config, callbacks=run_manager.get_child()),
                )
        except BaseException as e:
            await run_manager.on_chain_error(e)
            raise
        else:
            await run_manager.on_chain_end(input)
            return input

# 异步Web服务
from fastapi import FastAPI
app = FastAPI()

@app.post("/chat")
async def chat_endpoint(request: ChatRequest):
    # 使用ainvoke实现高并发
    result = await chain.ainvoke(
        {"messages": request.messages},
        config={"metadata": {"user_id": request.user_id}}
    )
    return {"response": result}

# 批量异步处理
results = await chain.abatch([input1, input2, input3])

sequenceDiagram
    autonumber
    participant User as 异步用户代码
    participant Seq as RunnableSequence
    participant Event as asyncio事件循环
    participant LLM as 异步LLM
    participant API as LLM API

    User->>Seq: await ainvoke(input)
    Seq->>Event: 注册coroutine
    Event->>Seq: 开始执行

    Seq->>LLM: await ainvoke(input)
    LLM->>API: httpx.post(url, ...) [async]
    Note over Event: 事件循环释放,<br/>可处理其他请求

    API-->>LLM: HTTP响应
    LLM-->>Seq: AIMessage
    Seq-->>Event: coroutine完成
    Event-->>User: final_output

retry_if_exception_type: tuple[type[BaseException], ...]  # 触发重试的异常类型
wait_exponential_jitter: bool  # 是否使用指数退避+抖动
stop_after_attempt: int  # 最大尝试次数

RunnableRetry[Input, Output]  # 包装后的Runnable

class Runnable(ABC, Generic[Input, Output]):
    def with_retry(
        self,
        *,
        retry_if_exception_type: tuple[type[BaseException], ...] = (Exception,),
        wait_exponential_jitter: bool = True,
        stop_after_attempt: int = 3,
    ) -> RunnableRetry[Input, Output]:
        from langchain_core.runnables.retry import RunnableRetry

        return RunnableRetry(
            bound=self,
            kwargs={},
            config={},
            retry_exception_types=retry_if_exception_type,
            wait_exponential_jitter=wait_exponential_jitter,
            max_attempt_number=stop_after_attempt,
        )

# RunnableRetry的实现
class RunnableRetry(RunnableBindingBase[Input, Output]):
    max_attempt_number: int = 3
    retry_exception_types: tuple[type[BaseException], ...] = (Exception,)
    wait_exponential_jitter: bool = True

    def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
        @retry(
            stop=stop_after_attempt(self.max_attempt_number),
            wait=wait_exponential_jitter() if self.wait_exponential_jitter else wait_exponential(),
            retry=retry_if_exception_type(self.retry_exception_types),
            before_sleep=self._before_sleep(config)
        )
        def _invoke_with_retry(input: Input, config: RunnableConfig) -> Output:
            return self.bound.invoke(input, config)

        return _invoke_with_retry(input, ensure_config(config))

# 为LLM添加重试
model_with_retry = ChatOpenAI(model="gpt-4o").with_retry(
    stop_after_attempt=5,
    retry_if_exception_type=(APIError, Timeout),
    wait_exponential_jitter=True
)

# 为整个链添加重试
chain_with_retry = (prompt | model | parser).with_retry(
    stop_after_attempt=3
)

# 调用(失败时自动重试)
result = chain_with_retry.invoke({"topic": "AI"})

sequenceDiagram
    autonumber
    participant User
    participant Retry as RunnableRetry
    participant Bound as 底层Runnable
    participant API as 外部API

    User->>Retry: invoke(input)

    Retry->>Bound: 第1次尝试
    Bound->>API: 请求
    API-->>Bound: 503 Service Unavailable
    Bound-->>Retry: APIError异常

    Note over Retry: 等待2^1秒(+jitter)

    Retry->>Bound: 第2次尝试
    Bound->>API: 请求
    API-->>Bound: 429 Rate Limit
    Bound-->>Retry: RateLimitError异常

    Note over Retry: 等待2^2秒(+jitter)

    Retry->>Bound: 第3次尝试
    Bound->>API: 请求
    API-->>Bound: 200 OK
    Bound-->>Retry: 成功结果

    Retry-->>User: 返回结果

fallbacks: Sequence[Runnable[Input, Output]]  # 备用Runnable列表
exceptions_to_handle: tuple[type[BaseException], ...]  # 触发故障转移的异常

RunnableWithFallbacks[Input, Output]  # 包装后的Runnable

class Runnable(ABC, Generic[Input, Output]):
    def with_fallbacks(
        self,
        fallbacks: Sequence[Runnable[Input, Output]],
        *,
        exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,),
    ) -> RunnableWithFallbacks[Input, Output]:
        from langchain_core.runnables.fallbacks import RunnableWithFallbacks

        return RunnableWithFallbacks(
            runnable=self,
            fallbacks=list(fallbacks),
            exceptions_to_handle=exceptions_to_handle,
        )

# RunnableWithFallbacks实现
class RunnableWithFallbacks(RunnableSerializable[Input, Output]):
    runnable: Runnable[Input, Output]
    fallbacks: list[Runnable[Input, Output]]
    exceptions_to_handle: tuple[type[BaseException], ...] = (Exception,)

    def invoke(self, input: Input, config: RunnableConfig | None = None) -> Output:
        config = ensure_config(config)
        callback_manager = get_callback_manager_for_config(config)
        run_manager = callback_manager.on_chain_start(dumpd(self), input)

        # 1) 尝试主Runnable
        try:
            return self.runnable.invoke(
                input,
                patch_config(config, callbacks=run_manager.get_child())
            )
        except self.exceptions_to_handle as e:
            # 2) 依次尝试备用Runnable
            for fallback in self.fallbacks:
                try:
                    return fallback.invoke(
                        input,
                        patch_config(config, callbacks=run_manager.get_child())
                    )
                except self.exceptions_to_handle:
                    continue

            # 3) 所有备用都失败,抛出原始异常
            run_manager.on_chain_error(e)
            raise
        else:
            run_manager.on_chain_end(output)
            return output

# 主模型+备用模型
primary_model = ChatOpenAI(model="gpt-4o", temperature=0)
fallback_model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

model_with_fallback = primary_model.with_fallbacks([fallback_model])

# 多级fallback
model_with_multi_fallback = primary_model.with_fallbacks([
    ChatOpenAI(model="gpt-4o-mini"),
    ChatOpenAI(model="gpt-3.5-turbo"),
])

# 整个链的fallback
chain_with_fallback = (prompt | model | parser).with_fallbacks([
    prompt | fallback_model | parser
])

sequenceDiagram
    autonumber
    participant User
    participant Fallbacks as RunnableWithFallbacks
    participant Primary as 主Runnable
    participant Fallback1 as 备用1
    participant Fallback2 as 备用2

    User->>Fallbacks: invoke(input)

    Fallbacks->>Primary: invoke(input)
    Primary->>Primary: 调用API
    Primary-->>Fallbacks: APIError异常
    Note over Fallbacks: 捕获异常,<br/>尝试备用1

    Fallbacks->>Fallback1: invoke(input)
    Fallback1->>Fallback1: 调用备用API
    Fallback1-->>Fallbacks: Timeout异常
    Note over Fallbacks: 继续尝试备用2

    Fallbacks->>Fallback2: invoke(input)
    Fallback2->>Fallback2: 调用另一备用API
    Fallback2-->>Fallbacks: 成功结果

    Fallbacks-->>User: 返回结果

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI

# 构建链
prompt = ChatPromptTemplate.from_template("讲一个关于{topic}的笑话")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser

# 调用
result = chain.invoke({"topic": "程序员"})