LangChain 源码剖析 - 架构图与时序图

graph TB
    subgraph "用户应用层"
        A1[RAG应用] 
        A2[聊天机器人]
        A3[代码生成器]
        A4[数据分析工具]
        A5[文档处理系统]
    end
    
    subgraph "LangChain框架层"
        B1[langchain] --> B11[chains - 链式处理]
        B1 --> B12[agents - 智能代理]
        B1 --> B13[memory - 记忆管理]
        B1 --> B14[retrievers - 检索器]
        
        B2[langchain-core] --> B21[runnables - 可运行接口]
        B2 --> B22[language_models - 语言模型]
        B2 --> B23[messages - 消息系统]
        B2 --> B24[prompts - 提示模板]
        B2 --> B25[tools - 工具系统]
        B2 --> B26[callbacks - 回调系统]
        B2 --> B27[vectorstores - 向量存储]
        
        B3[langchain-text-splitters] --> B31[文本分割算法]
    end
    
    subgraph "合作伙伴集成层"
        C1[langchain-openai] --> C11[OpenAI模型]
        C2[langchain-anthropic] --> C21[Claude模型]
        C3[langchain-chroma] --> C31[Chroma向量库]
        C4[langchain-pinecone] --> C41[Pinecone向量库]
        C5[其他集成包...] --> C51[各种模型和工具]
    end
    
    subgraph "外部服务层"
        D1[OpenAI API]
        D2[Anthropic API]
        D3[Hugging Face]
        D4[向量数据库]
        D5[搜索引擎]
        D6[文件系统]
        D7[数据库]
    end
    
    subgraph "开发运维层"
        E1[LangSmith] --> E11[调试追踪]
        E1 --> E12[性能监控]
        E1 --> E13[评估测试]
        
        E2[LangGraph] --> E21[复杂工作流]
        E2 --> E22[Agent编排]
        E2 --> E23[状态管理]
    end
    
    %% 依赖关系
    A1 --> B1
    A2 --> B1
    A3 --> B1
    A4 --> B1
    A5 --> B1
    
    B1 --> B2
    B1 --> B3
    B1 --> C1
    B1 --> C2
    B1 --> C3
    B1 --> C4
    
    C1 --> D1
    C2 --> D2
    C3 --> D4
    C4 --> D4
    C5 --> D3
    C5 --> D5
    C5 --> D6
    C5 --> D7
    
    B1 -.-> E1
    B1 -.-> E2

graph LR
    subgraph "LangChain Core 核心抽象层"
        A[Runnable接口] --> A1[统一执行协议]
        A --> A2[组合操作符]
        A --> A3[配置管理]
        
        B[消息系统] --> B1[BaseMessage]
        B --> B2[多模态内容]
        B --> B3[工具调用]
        
        C[语言模型抽象] --> C1[BaseChatModel]
        C --> C2[BaseLLM]
        C --> C3[嵌入模型]
        
        D[工具系统] --> D1[BaseTool]
        D --> D2[@tool装饰器]
        D --> D3[工具调用协议]
        
        E[提示系统] --> E1[PromptTemplate]
        E --> E2[ChatPromptTemplate]
        E --> E3[少样本学习]
        
        F[向量存储] --> F1[VectorStore抽象]
        F --> F2[检索器接口]
        F --> F3[相似性搜索]
        
        G[回调系统] --> G1[执行追踪]
        G --> G2[性能监控]
        G --> G3[错误处理]
    end
    
    subgraph "LangChain 应用层"
        H[链系统] --> H1[RunnableSequence]
        H --> H2[RunnableParallel]
        H --> H3[专用链类型]
        
        I[Agent系统] --> I1[ReAct Agent]
        I --> I2[工具调用Agent]
        I --> I3[计划执行Agent]
        
        J[检索系统] --> J1[文档加载器]
        J --> J2[文本分割器]
        J --> J3[RAG链]
        
        K[记忆系统] --> K1[对话缓冲区]
        K --> K2[向量记忆]
        K --> K3[总结记忆]
    end
    
    %% 依赖关系
    H --> A
    H --> B
    H --> C
    I --> A
    I --> D
    I --> B
    J --> F
    J --> E
    K --> F
    K --> B

sequenceDiagram
    participant User as 用户应用
    participant Chain as 链对象
    participant Config as 配置管理器
    participant Callback as 回调管理器
    participant Component1 as 组件1
    participant Component2 as 组件2
    participant LLM as 语言模型
    
    User->>Chain: invoke(input, config)
    Chain->>Config: ensure_config(config)
    Config-->>Chain: validated_config
    
    Chain->>Callback: get_callback_manager_for_config()
    Callback-->>Chain: callback_manager
    
    Chain->>Callback: on_chain_start(serialized, inputs)
    
    Note over Chain: 开始执行链式处理
    
    Chain->>Component1: invoke(input, step_config)
    Component1->>Callback: on_component_start()
    
    Component1->>LLM: _generate(messages)
    LLM->>Callback: on_llm_start(serialized, prompts)
    LLM-->>Component1: result
    LLM->>Callback: on_llm_end(response)
    
    Component1-->>Chain: intermediate_result
    Component1->>Callback: on_component_end()
    
    Chain->>Component2: invoke(intermediate_result, step_config)
    Component2->>Callback: on_component_start()
    Component2-->>Chain: final_result
    Component2->>Callback: on_component_end()
    
    Chain->>Callback: on_chain_end(outputs)
    Chain-->>User: final_result

sequenceDiagram
    participant User as 用户应用
    participant Chain as 流式链
    participant StreamingLLM as 流式语言模型
    participant Callback as 回调管理器
    participant TokenHandler as Token处理器
    
    User->>Chain: stream(input, config)
    Chain->>Callback: on_chain_start()
    
    Chain->>StreamingLLM: _stream(messages)
    StreamingLLM->>Callback: on_llm_start()
    
    loop 流式生成过程
        StreamingLLM-->>Chain: message_chunk
        StreamingLLM->>Callback: on_llm_new_token(token)
        Chain->>TokenHandler: process_chunk(chunk)
        TokenHandler-->>Chain: processed_chunk
        Chain-->>User: yield processed_chunk
    end
    
    StreamingLLM->>Callback: on_llm_end(final_result)
    Chain->>Callback: on_chain_end(aggregated_result)

sequenceDiagram
    participant User as 用户
    participant Agent as Agent执行器
    participant LLM as 语言模型
    participant ToolManager as 工具管理器
    participant Tool1 as 搜索工具
    participant Tool2 as 计算工具
    participant Callback as 回调系统
    
    User->>Agent: invoke("帮我搜索天气并计算温度转换")
    Agent->>Callback: on_chain_start()
    
    Note over Agent: 第一轮思考
    Agent->>LLM: generate_with_tools(query, available_tools)
    LLM->>Callback: on_llm_start()
    LLM-->>Agent: AIMessage(tool_calls=[search_weather])
    LLM->>Callback: on_llm_end()
    
    Agent->>Callback: on_agent_action(action)
    Agent->>ToolManager: execute_tool_call(tool_call)
    ToolManager->>Tool1: invoke("北京天气")
    Tool1->>Callback: on_tool_start()
    Tool1-->>ToolManager: "北京：晴天，15°C"
    Tool1->>Callback: on_tool_end()
    ToolManager-->>Agent: tool_result
    
    Note over Agent: 第二轮思考
    Agent->>LLM: continue_with_tool_result(tool_result)
    LLM->>Callback: on_llm_start()
    LLM-->>Agent: AIMessage(tool_calls=[calculate])
    LLM->>Callback: on_llm_end()
    
    Agent->>ToolManager: execute_tool_call(calculate_call)
    ToolManager->>Tool2: invoke("15°C转华氏度")
    Tool2->>Callback: on_tool_start()
    Tool2-->>ToolManager: "59°F"
    Tool2->>Callback: on_tool_end()
    ToolManager-->>Agent: calculation_result
    
    Note over Agent: 最终回答
    Agent->>LLM: finalize_answer(all_results)
    LLM->>Callback: on_llm_start()
    LLM-->>Agent: AIMessage("北京今天晴天，15°C (59°F)")
    LLM->>Callback: on_llm_end()
    
    Agent->>Callback: on_agent_finish(final_answer)
    Agent->>Callback: on_chain_end()
    Agent-->>User: "北京今天晴天，15°C (59°F)"

sequenceDiagram
    participant User as 用户
    participant RAGChain as RAG链
    participant Retriever as 检索器
    participant VectorStore as 向量存储
    participant Embeddings as 嵌入模型
    participant LLM as 生成模型
    participant Callback as 回调系统
    
    User->>RAGChain: invoke("什么是LangChain？")
    RAGChain->>Callback: on_chain_start()
    
    Note over RAGChain: 检索阶段
    RAGChain->>Retriever: get_relevant_documents(query)
    Retriever->>Callback: on_retriever_start()
    
    Retriever->>Embeddings: embed_query("什么是LangChain？")
    Embeddings->>Callback: on_llm_start()
    Embeddings-->>Retriever: query_vector
    Embeddings->>Callback: on_llm_end()
    
    Retriever->>VectorStore: similarity_search(query_vector, k=4)
    VectorStore-->>Retriever: relevant_documents
    Retriever->>Callback: on_retriever_end(documents)
    Retriever-->>RAGChain: documents
    
    Note over RAGChain: 上下文构建
    RAGChain->>RAGChain: build_context(documents, query)
    
    Note over RAGChain: 生成阶段
    RAGChain->>LLM: generate(context + query)
    LLM->>Callback: on_llm_start()
    
    opt 如果支持流式输出
        loop 流式生成
            LLM-->>RAGChain: chunk
            LLM->>Callback: on_llm_new_token()
            RAGChain-->>User: yield chunk
        end
    else 完整输出
        LLM-->>RAGChain: complete_answer
    end
    
    LLM->>Callback: on_llm_end()
    RAGChain->>Callback: on_chain_end()
    RAGChain-->>User: final_answer

flowchart TD
    A[开始执行Runnable] --> B{配置检查}
    B -->|配置无效| B1[ConfigError]
    B -->|配置有效| C[初始化回调管理器]
    
    C --> D[开始执行核心逻辑]
    D --> E{执行成功?}
    
    E -->|成功| F[触发成功回调]
    F --> G[返回结果]
    
    E -->|失败| H{异常类型判断}
    H -->|ValidationError| H1[参数验证错误]
    H -->|TimeoutError| H2[执行超时]
    H -->|APIError| H3[外部API错误]
    H -->|ToolException| H4[工具执行错误]
    H -->|其他异常| H5[未知错误]
    
    H1 --> I{配置了错误处理?}
    H2 --> I
    H3 --> I
    H4 --> I
    H5 --> I
    
    I -->|是| J[执行错误处理策略]
    I -->|否| K[触发错误回调]
    
    J --> J1{处理策略类型}
    J1 -->|重试| L[重试机制]
    J1 -->|回退| M[回退机制]
    J1 -->|忽略| N[返回默认值]
    J1 -->|转换| O[转换为成功结果]
    
    L --> L1{重试次数检查}
    L1 -->|未达上限| L2[等待重试间隔]
    L2 --> D
    L1 -->|达到上限| P[重试失败]
    
    M --> M1{有备用方案?}
    M1 -->|有| M2[执行备用Runnable]
    M2 --> E
    M1 -->|无| P
    
    N --> F
    O --> F
    P --> K
    K --> Q[抛出异常]

sequenceDiagram
    participant User as 用户
    participant RetryWrapper as 重试包装器
    participant OriginalRunnable as 原始Runnable
    participant BackoffStrategy as 退避策略
    participant Callback as 回调系统
    
    User->>RetryWrapper: invoke(input)
    RetryWrapper->>Callback: on_retry_start()
    
    Note over RetryWrapper: 第1次尝试
    RetryWrapper->>OriginalRunnable: invoke(input)
    OriginalRunnable-->>RetryWrapper: Exception("API限流")
    RetryWrapper->>Callback: on_retry_attempt(attempt=1, error)
    
    RetryWrapper->>BackoffStrategy: calculate_wait_time(attempt=1)
    BackoffStrategy-->>RetryWrapper: wait_time=2s
    Note over RetryWrapper: 等待 2 秒
    
    Note over RetryWrapper: 第2次尝试
    RetryWrapper->>OriginalRunnable: invoke(input)
    OriginalRunnable-->>RetryWrapper: Exception("网络超时")
    RetryWrapper->>Callback: on_retry_attempt(attempt=2, error)
    
    RetryWrapper->>BackoffStrategy: calculate_wait_time(attempt=2)
    BackoffStrategy-->>RetryWrapper: wait_time=4s
    Note over RetryWrapper: 等待 4 秒
    
    Note over RetryWrapper: 第3次尝试
    RetryWrapper->>OriginalRunnable: invoke(input)
    OriginalRunnable-->>RetryWrapper: success_result
    RetryWrapper->>Callback: on_retry_success(attempt=3)
    
    RetryWrapper-->>User: success_result

flowchart TD
    A[批处理请求] --> B[输入验证和预处理]
    B --> C{配置并发限制}
    
    C --> D[创建任务组]
    D --> E[分批处理]
    
    E --> F[ThreadPoolExecutor]
    F --> G1[Worker线程1]
    F --> G2[Worker线程2]
    F --> G3[Worker线程N]
    
    G1 --> H1[执行Runnable实例1]
    G2 --> H2[执行Runnable实例2]
    G3 --> H3[执行Runnable实例N]
    
    H1 --> I1{执行结果}
    H2 --> I2{执行结果}
    H3 --> I3{执行结果}
    
    I1 -->|成功| J1[收集结果1]
    I1 -->|失败| K1[收集错误1]
    I2 -->|成功| J2[收集结果2]
    I2 -->|失败| K2[收集错误2]
    I3 -->|成功| J3[收集结果N]
    I3 -->|失败| K3[收集错误N]
    
    J1 --> L[结果聚合]
    J2 --> L
    J3 --> L
    K1 --> L
    K2 --> L
    K3 --> L
    
    L --> M[返回批处理结果]

sequenceDiagram
    participant User as 用户线程
    participant AsyncExecutor as 异步执行器
    participant EventLoop as 事件循环
    participant Task1 as 异步任务1
    participant Task2 as 异步任务2
    participant Task3 as 异步任务3
    participant External as 外部服务
    
    User->>AsyncExecutor: abatch([input1, input2, input3])
    AsyncExecutor->>EventLoop: 创建事件循环
    
    par 并行任务创建
        AsyncExecutor->>Task1: create_task(ainvoke(input1))
        AsyncExecutor->>Task2: create_task(ainvoke(input2))
        AsyncExecutor->>Task3: create_task(ainvoke(input3))
    end
    
    Note over EventLoop: 并发执行异步任务
    
    par 并行外部调用
        Task1->>External: API调用1
        Task2->>External: API调用2
        Task3->>External: API调用3
    end
    
    par 并行结果处理
        External-->>Task1: 响应1
        External-->>Task2: 响应2
        External-->>Task3: 响应3
    end
    
    par 任务完成
        Task1-->>EventLoop: result1
        Task2-->>EventLoop: result2
        Task3-->>EventLoop: result3
    end
    
    EventLoop->>AsyncExecutor: gather(results)
    AsyncExecutor-->>User: [result1, result2, result3]

graph TD
    subgraph "缓存层级架构"
        A[用户请求] --> B{缓存检查}
        
        B -->|命中| B1[内存缓存]
        B1 --> B2[返回缓存结果]
        
        B -->|未命中内存| C{持久化缓存检查}
        C -->|命中| C1[SQLite缓存]
        C1 --> C2[加载到内存]
        C2 --> B2
        
        C -->|未命中| D[执行LLM调用]
        D --> D1[获取结果]
        D1 --> D2[写入缓存]
        D2 --> D3[返回结果]
        
        subgraph "缓存策略"
            E1[LRU淘汰]
            E2[TTL过期]
            E3[大小限制]
        end
        
        B1 -.-> E1
        C1 -.-> E2
        D2 -.-> E3
    end
    
    subgraph "缓存键生成"
        F[请求参数] --> F1[提示文本]
        F --> F2[模型参数]
        F --> F3[停止词]
        F1 --> G[哈希计算]
        F2 --> G
        F3 --> G
        G --> H[缓存键]
    end

sequenceDiagram
    participant User as 用户
    participant LLM as 语言模型
    participant CacheManager as 缓存管理器
    participant MemoryCache as 内存缓存
    participant PersistentCache as 持久化缓存
    
    User->>LLM: generate("什么是AI？")
    LLM->>CacheManager: check_cache(prompt_hash)
    
    CacheManager->>MemoryCache: get(cache_key)
    alt 内存缓存命中
        MemoryCache-->>CacheManager: cached_result
        CacheManager-->>LLM: cached_result
        LLM-->>User: cached_result
    else 内存缓存未命中
        MemoryCache-->>CacheManager: None
        CacheManager->>PersistentCache: get(cache_key)
        
        alt 持久化缓存命中
            PersistentCache-->>CacheManager: cached_result
            CacheManager->>MemoryCache: put(cache_key, result)
            CacheManager-->>LLM: cached_result
            LLM-->>User: cached_result
        else 完全未命中
            PersistentCache-->>CacheManager: None
            CacheManager-->>LLM: cache_miss
            
            LLM->>LLM: 执行实际生成
            LLM->>CacheManager: store_result(cache_key, result)
            CacheManager->>MemoryCache: put(cache_key, result)
            CacheManager->>PersistentCache: put(cache_key, result)
            LLM-->>User: fresh_result
        end
    end

graph TB
    subgraph "LangChain应用"
        A[用户应用] --> B[LangChain组件]
        B --> C[自动追踪]
    end
    
    subgraph "回调系统"
        C --> D[LangSmithCallbackHandler]
        D --> E[运行数据收集]
        E --> F[性能指标计算]
        F --> G[错误信息捕获]
    end
    
    subgraph "数据传输"
        G --> H[批量数据]
        H --> I[压缩和序列化]
        I --> J[HTTP客户端]
        J --> K[异步上传]
    end
    
    subgraph "LangSmith平台"
        K --> L[数据接收API]
        L --> M[数据存储]
        M --> N[索引和查询]
        N --> O[可视化界面]
        N --> P[分析和报表]
    end
    
    subgraph "用户界面"
        O --> Q[执行追踪查看器]
        P --> R[性能dashboard]
        P --> S[错误监控]
        P --> T[A/B测试结果]
    end

sequenceDiagram
    participant App as 应用
    participant Component as LangChain组件
    participant Tracer as 追踪器
    participant Collector as 数据收集器
    participant Buffer as 缓冲区
    participant Uploader as 上传器
    participant LangSmith as LangSmith服务
    
    App->>Component: 执行操作
    Component->>Tracer: on_component_start()
    Tracer->>Collector: 记录开始事件
    
    Note over Component: 执行具体逻辑
    
    Component->>Tracer: on_llm_start()
    Tracer->>Collector: 记录LLM调用
    Component->>Tracer: on_llm_new_token()
    Tracer->>Collector: 记录token生成
    Component->>Tracer: on_llm_end()
    Tracer->>Collector: 记录LLM完成
    
    Component->>Tracer: on_component_end()
    Tracer->>Collector: 记录完成事件
    
    Collector->>Buffer: 存储运行数据
    
    Note over Buffer: 达到批次大小或时间间隔
    
    Buffer->>Uploader: 触发上传
    Uploader->>LangSmith: POST /runs/batch
    LangSmith-->>Uploader: 上传确认
    Uploader->>Buffer: 清理已上传数据

flowchart TD
    A[用户输入] --> B[输入验证器]
    B --> B1{格式检查}
    B1 -->|无效| B2[拒绝请求]
    B1 -->|有效| C[内容安全检查]
    
    C --> C1{危险模式检测}
    C1 -->|发现危险| C2[阻断执行]
    C1 -->|安全| D[权限检查]
    
    D --> D1{用户权限}
    D1 -->|无权限| D2[权限拒绝]
    D1 -->|有权限| E[速率限制检查]
    
    E --> E1{请求频率}
    E1 -->|超限| E2[限流响应]
    E1 -->|正常| F[执行LangChain组件]
    
    F --> G[输出过滤器]
    G --> G1{敏感信息检测}
    G1 -->|发现敏感信息| G2[信息脱敏]
    G1 -->|安全| H[返回结果]
    G2 --> H
    
    B2 --> I[记录安全事件]
    C2 --> I
    D2 --> I
    E2 --> I

sequenceDiagram
    participant User as 用户
    participant AuthMiddleware as 认证中间件
    participant PermissionChecker as 权限检查器
    participant SecurityScanner as 安全扫描器
    participant LangChainApp as LangChain应用
    participant AuditLogger as 审计日志
    
    User->>AuthMiddleware: 请求执行
    AuthMiddleware->>AuthMiddleware: 验证身份
    
    alt 身份验证失败
        AuthMiddleware->>AuditLogger: 记录验证失败
        AuthMiddleware-->>User: 401 Unauthorized
    else 身份验证成功
        AuthMiddleware->>PermissionChecker: check_permissions(user, action)
        
        alt 权限检查失败
            PermissionChecker->>AuditLogger: 记录权限拒绝
            PermissionChecker-->>User: 403 Forbidden
        else 权限检查通过
            PermissionChecker->>SecurityScanner: scan_input(request)
            
            alt 发现安全风险
                SecurityScanner->>AuditLogger: 记录安全风险
                SecurityScanner-->>User: 400 Bad Request
            else 安全检查通过
                SecurityScanner->>LangChainApp: execute(request)
                LangChainApp->>AuditLogger: 记录操作开始
                
                LangChainApp-->>SecurityScanner: result
                SecurityScanner->>SecurityScanner: scan_output(result)
                SecurityScanner-->>PermissionChecker: filtered_result
                PermissionChecker-->>AuthMiddleware: final_result
                AuthMiddleware->>AuditLogger: 记录操作完成
                AuthMiddleware-->>User: final_result
            end
        end
    end