flowchart TB
    subgraph Application["应用层"]
        APP[用户应用]
        AGENT[Agent应用]
        TOOL_APP[Tool应用]
    end

    subgraph LCEL["LCEL层 - 表达式语言"]
        SEQ[RunnableSequence<br/>顺序执行]
        PAR[RunnableParallel<br/>并行执行]
        BRANCH[RunnableBranch<br/>条件分支]
        LAMBDA[RunnableLambda<br/>自定义逻辑]
    end

    subgraph Core["langchain-core 核心抽象层"]
        RUNNABLE[Runnable<br/>统一执行协议]
        LM[BaseLanguageModel<br/>语言模型抽象]
        CHAT[BaseChatModel<br/>聊天模型]
        LLM[BaseLLM<br/>文本模型]
        MSG[Messages<br/>消息体系]
        PROMPT[Prompts<br/>提示模板]
        PARSER[OutputParsers<br/>输出解析]
        TOOL[Tools<br/>工具抽象]
        VECTOR[VectorStores<br/>向量存储]
        RETRIEVER[Retrievers<br/>检索器]
        CALLBACK[Callbacks<br/>回调系统]
    end

    subgraph Integrations["集成层"]
        PARTNERS[Partner包<br/>openai/anthropic/etc]
        CLASSIC[langchain-classic<br/>经典实现]
        SPLITTER[text-splitters<br/>文本分割]
    end

    subgraph External["外部系统"]
        LLM_API[LLM API<br/>OpenAI/Anthropic]
        VECTOR_DB[向量数据库<br/>Chroma/Pinecone]
        TOOLS_EXT[外部工具<br/>Search/APIs]
        LANGSMITH[LangSmith<br/>追踪监控]
    end

    APP --> LCEL
    AGENT --> LCEL
    TOOL_APP --> Core

    LCEL --> RUNNABLE
    SEQ -.继承.-> RUNNABLE
    PAR -.继承.-> RUNNABLE
    BRANCH -.继承.-> RUNNABLE
    LAMBDA -.继承.-> RUNNABLE

    RUNNABLE --> LM
    RUNNABLE --> PROMPT
    RUNNABLE --> PARSER
    RUNNABLE --> TOOL
    RUNNABLE --> VECTOR

    LM --> CHAT
    LM --> LLM
    CHAT --> MSG
    PROMPT --> MSG

    Core --> Integrations
    CHAT -.实现.-> PARTNERS
    LLM -.实现.-> PARTNERS
    VECTOR -.实现.-> PARTNERS
    TOOL -.实现.-> CLASSIC

    PARTNERS --> LLM_API
    PARTNERS --> VECTOR_DB
    CLASSIC --> TOOLS_EXT

    CALLBACK --> LANGSMITH

用户调用 → LCEL构建链 → Runnable.invoke → 组件执行 → 回调追踪 → 返回结果

输入数据 → Prompt格式化 → LLM推理 → OutputParser解析 → 结构化输出

sequenceDiagram
    autonumber
    participant User as 用户代码
    participant LCEL as LCEL链
    participant Runnable as Runnable基类
    participant Prompt as PromptTemplate
    participant LLM as ChatModel
    participant Parser as OutputParser
    participant CB as CallbackManager
    participant Tracer as LangSmith

    User->>LCEL: chain.invoke(input)
    LCEL->>Runnable: 解析执行计划

    Runnable->>CB: 创建回调管理器
    CB->>Tracer: 开始追踪(run_id)

    Runnable->>Prompt: invoke(input, config)
    Prompt->>Prompt: 格式化模板
    Prompt-->>Runnable: messages

    Runnable->>LLM: invoke(messages, config)
    LLM->>CB: on_llm_start
    CB->>Tracer: 记录输入

    LLM->>LLM: 调用API(OpenAI等)
    Note over LLM: 网络请求<br/>Token计数<br/>速率限制

    LLM-->>Runnable: AIMessage
    LLM->>CB: on_llm_end
    CB->>Tracer: 记录输出和Token

    Runnable->>Parser: invoke(message, config)
    Parser->>Parser: 解析结构化输出
    Parser-->>Runnable: parsed_output

    Runnable->>CB: on_chain_end
    CB->>Tracer: 结束追踪
    Runnable-->>User: final_output

# 批处理使用线程池
with ThreadPoolExecutor(max_workers=config.max_workers) as executor:
    futures = [executor.submit(self.invoke, input) for input in inputs]

# 异步使用gather_with_concurrency
async def abatch(self, inputs):
    tasks = [self.ainvoke(input) for input in inputs]
    return await gather_with_concurrency(config.max_concurrency, *tasks)

{
    "callbacks": [...],           # 回调处理器列表
    "tags": ["env:prod"],        # 标签,用于过滤和分组
    "metadata": {"user_id": "123"}, # 元数据,记录到trace
    "run_name": "my_chain",      # 运行名称
    "max_concurrency": 5,        # 最大并发数
    "recursion_limit": 25,       # 递归深度限制
    "timeout": 60,               # 超时(秒)
    "configurable": {...}        # 动态配置字段
}

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

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

chain = prompt | model | parser

# 调用
result = chain.invoke({"topic": "程序员"})
print(result)  # "为什么程序员总是混淆圣诞节和万圣节?因为 Oct 31 == Dec 25!"

from langchain_core.runnables import RunnableParallel
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class Joke(BaseModel):
    setup: str = Field(description="笑话的铺垫")
    punchline: str = Field(description="笑点")

# 并行生成多个字段
chain = RunnableParallel({
    "joke": prompt | model | JsonOutputParser(pydantic_object=Joke),
    "topic_analysis": ChatPromptTemplate.from_template(
        "分析主题'{topic}'的幽默潜力"
    ) | model | StrOutputParser()
})

result = chain.invoke({"topic": "AI"})
# {
#   "joke": {"setup": "...", "punchline": "..."},
#   "topic_analysis": "AI作为主题..."
# }

from langchain_core.runnables import RunnableWithFallbacks

# 主模型
primary = ChatOpenAI(model="gpt-4o", temperature=0)

# 备用模型
fallback = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 配置故障转移
model_with_fallbacks = primary.with_fallbacks([fallback])

# 配置重试(指数退避)
model_with_retry = model_with_fallbacks.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True
)

chain = prompt | model_with_retry | parser

# 当gpt-4o失败时自动切换到gpt-4o-mini
result = chain.invoke({"topic": "量子计算"})

# 流式处理
for chunk in chain.stream({"topic": "太空"}):
    print(chunk, end="", flush=True)
# 输出逐字显示...

# 异步流式
async def async_stream():
    async for chunk in chain.astream({"topic": "深海"}):
        print(chunk, end="", flush=True)

import asyncio
asyncio.run(async_stream())

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个并发
)

from langchain_core.runnables import Runnable

class MyCustomRunnable(Runnable[str, str]):
    def invoke(self, input: str, config: RunnableConfig = None) -> str:
        # 实现同步逻辑
        return input.upper()

    async def ainvoke(self, input: str, config: RunnableConfig = None) -> str:
        # 实现异步逻辑(可选,默认调用invoke)
        return self.invoke(input, config)