LangChain完整源码分析总览

graph TD
    A[框架使用手册] --> B[架构概览]
    B --> C[Core模块基础]
    C --> D[Chain模块入门]
    D --> E[简单应用实践]

    A --> A1[快速入门]
    A --> A2[核心概念]
    A --> A3[基本用法]

    B --> B1[分层架构]
    B --> B2[核心抽象]
    B --> B3[设计模式]

    C --> C1[Runnable接口]
    C --> C2[基础组件]
    C --> C3[配置系统]

    D --> D1[链式调用]
    D --> D2[组合模式]
    D --> D3[实际案例]

graph TD
    A[Agent模块深入] --> B[工具系统]
    B --> C[内存管理]
    C --> D[合作伙伴集成]
    D --> E[性能优化]

    A --> A1[Agent架构]
    A --> A2[执行器模式]
    A --> A3[工具调用]

    B --> B1[工具抽象]
    B --> B2[自定义工具]
    B --> B3[工具组合]

    C --> C1[内存类型]
    C --> C2[向量存储]
    C --> C3[检索策略]

    D --> D1[LLM集成]
    D --> D2[向量数据库]
    D --> D3[第三方服务]

    E --> E1[缓存策略]
    E --> E2[批量处理]
    E --> E3[资源管理]

graph TD
    A[高级架构设计] --> B[企业级实践]
    B --> C[生产部署]
    C --> D[监控运维]
    D --> E[问题诊断]

    A --> A1[微服务架构]
    A --> A2[分布式系统]
    A --> A3[扩展性设计]

    B --> B1[安全策略]
    B --> B2[合规要求]
    B --> B3[成本优化]

    C --> C1[容器化部署]
    C --> C2[CI/CD流程]
    C --> C3[蓝绿部署]

    D --> D1[指标监控]
    D --> D2[日志分析]
    D --> D3[告警系统]

    E --> E1[性能调优]
    E --> E2[故障排查]
    E --> E3[容量规划]

graph TB
    subgraph "应用层 (Application Layer)"
        A1[Web API]
        A2[CLI工具]
        A3[Jupyter Notebook]
    end

    subgraph "业务层 (Business Layer)"
        B1[Agent服务]
        B2[Chain服务]
        B3[RAG服务]
        B4[工具服务]
    end

    subgraph "核心层 (Core Layer)"
        C1[Runnable接口]
        C2[BaseLanguageModel]
        C3[BaseTool]
        C4[BaseMemory]
        C5[BaseCallbackHandler]
    end

    subgraph "集成层 (Integration Layer)"
        D1[LLM适配器]
        D2[向量存储适配器]
        D3[工具适配器]
        D4[外部API适配器]
    end

    subgraph "基础设施层 (Infrastructure Layer)"
        E1[OpenAI/Anthropic]
        E2[Pinecone/Weaviate]
        E3[数据库/缓存]
        E4[监控/日志]
    end

    A1 --> B1
    A2 --> B2
    A3 --> B3

    B1 --> C1
    B2 --> C2
    B3 --> C3
    B4 --> C4

    C1 --> D1
    C2 --> D2
    C3 --> D3
    C4 --> D4

    D1 --> E1
    D2 --> E2
    D3 --> E3
    D4 --> E4

classDiagram
    class Runnable {
        +invoke(input, config)
        +ainvoke(input, config)
        +stream(input, config)
        +astream(input, config)
        +batch(inputs, config)
        +abatch(inputs, config)
    }

    class RunnableSequence {
        +first: Runnable
        +middle: List[Runnable]
        +last: Runnable
        +invoke(input, config)
    }

    class RunnableParallel {
        +steps: Dict[str, Runnable]
        +invoke(input, config)
    }

    class BaseLanguageModel {
        +generate(prompts, stop, callbacks)
        +agenerate(prompts, stop, callbacks)
        +predict(text, stop, callbacks)
        +apredict(text, stop, callbacks)
    }

    class BaseChatModel {
        +generate(messages, stop, callbacks)
        +agenerate(messages, stop, callbacks)
        +predict_messages(messages, stop, callbacks)
    }

    class BaseTool {
        +name: str
        +description: str
        +run(tool_input, verbose, start_color, color, callbacks)
        +arun(tool_input, verbose, start_color, color, callbacks)
    }

    class BaseMemory {
        +memory_variables: List[str]
        +load_memory_variables(inputs)
        +save_context(inputs, outputs)
        +clear()
    }

    Runnable <|-- RunnableSequence
    Runnable <|-- RunnableParallel
    Runnable <|-- BaseLanguageModel
    BaseLanguageModel <|-- BaseChatModel
    Runnable <|-- BaseTool
    Runnable <|-- BaseMemory

# langchain_core/runnables/base.py
class Runnable(Generic[Input, Output], ABC):
    """
    所有可运行组件的基础抽象类

    核心设计理念：
    1. 统一的调用接口 - invoke/ainvoke
    2. 流式处理支持 - stream/astream
    3. 批量处理能力 - batch/abatch
    4. 组合能力 - 通过 | 和 & 操作符
    """

    @abstractmethod
    def invoke(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None
    ) -> Output:
        """同步调用 - 核心执行方法"""
        pass

    async def ainvoke(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None
    ) -> Output:
        """异步调用 - 默认实现调用同步版本"""
        return await asyncio.get_event_loop().run_in_executor(
            None, partial(self.invoke, input, config=config)
        )

    def stream(
        self,
        input: Input,
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Iterator[Output]:
        """流式输出 - 默认实现返回完整结果"""
        yield self.invoke(input, config, **kwargs)

    def batch(
        self,
        inputs: List[Input],
        config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None,
        *,
        return_exceptions: bool = False,
        **kwargs: Any,
    ) -> List[Output]:
        """批量处理 - 默认实现逐个调用"""
        if not inputs:
            return []

        # 处理配置
        configs = self._get_config_list(config, len(inputs))

        # 批量执行
        outputs = []
        for i, input_item in enumerate(inputs):
            try:
                output = self.invoke(input_item, configs[i], **kwargs)
                outputs.append(output)
            except Exception as e:
                if return_exceptions:
                    outputs.append(e)
                else:
                    raise e

        return outputs

    def __or__(self, other: Runnable) -> RunnableSequence:
        """管道操作符 - 创建序列"""
        return RunnableSequence(first=self, last=other)

    def __and__(self, other: Runnable) -> RunnableParallel:
        """并行操作符 - 创建并行执行"""
        return RunnableParallel(steps={"left": self, "right": other})

# langchain/chains/base.py
class Chain(Runnable[Dict[str, Any], Dict[str, Any]], ABC):
    """
    链式执行的基础类

    设计模式：
    1. 模板方法模式 - _call方法定义执行框架
    2. 策略模式 - 不同链类型实现不同策略
    3. 装饰器模式 - 回调和内存装饰核心逻辑
    """

    memory: Optional[BaseMemory] = None
    callbacks: Callbacks = Field(default=None, exclude=True)
    callback_manager: Optional[BaseCallbackManager] = Field(default=None, exclude=True)
    verbose: bool = Field(default_factory=_get_verbosity)
    tags: Optional[List[str]] = None
    metadata: Optional[Dict[str, Any]] = None

    @property
    @abstractmethod
    def input_keys(self) -> List[str]:
        """输入键列表 - 定义链的输入接口"""
        pass

    @property
    @abstractmethod
    def output_keys(self) -> List[str]:
        """输出键列表 - 定义链的输出接口"""
        pass

    @abstractmethod
    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """核心执行逻辑 - 子类必须实现"""
        pass

    def invoke(
        self,
        input: Dict[str, Any],
        config: Optional[RunnableConfig] = None,
    ) -> Dict[str, Any]:
        """
        统一调用入口 - 实现完整的执行流程

        执行流程：
        1. 输入验证和预处理
        2. 内存加载（如果有）
        3. 回调管理器设置
        4. 核心逻辑执行
        5. 内存保存（如果有）
        6. 结果后处理
        """
        config = ensure_config(config)

        # 1. 输入验证
        inputs = self.prep_inputs(input)

        # 2. 设置回调管理器
        callback_manager = CallbackManager.configure(
            config.get("callbacks"),
            self.callbacks,
            self.verbose,
            config.get("tags"),
            self.tags,
            config.get("metadata"),
            self.metadata,
        )

        # 3. 执行链
        run_manager = callback_manager.on_chain_start(
            dumpd(self),
            inputs,
            name=config.get("run_name"),
        )

        try:
            # 4. 核心执行
            outputs = self._call(inputs, run_manager)

            # 5. 保存到内存
            if self.memory is not None:
                self.memory.save_context(inputs, outputs)

            # 6. 回调结束
            run_manager.on_chain_end(outputs)

            return outputs

        except Exception as e:
            run_manager.on_chain_error(e)
            raise e

    def prep_inputs(self, inputs: Union[Dict[str, Any], Any]) -> Dict[str, Any]:
        """输入预处理和验证"""
        if not isinstance(inputs, dict):
            _input_keys = set(self.input_keys)
            if self.memory is not None:
                # 如果有内存，从内存加载变量
                _input_keys = _input_keys.difference(self.memory.memory_variables)

            if len(_input_keys) != 1:
                raise ValueError(
                    f"A single string input was passed in, but this chain expects "
                    f"multiple inputs ({_input_keys}). When a chain expects "
                    f"multiple inputs, please call it by passing in a dictionary, "
                    f"eg `chain({'key1': 'value1', 'key2': 'value2'})`"
                )

            inputs = {list(_input_keys)[0]: inputs}

        # 从内存加载变量
        if self.memory is not None:
            external_context = self.memory.load_memory_variables(inputs)
            inputs = dict(inputs, **external_context)

        # 验证输入键
        self._validate_inputs(inputs)
        return inputs

# langchain/agents/agent.py
class BaseMultiActionAgent(BaseAgent):
    """
    多动作Agent基类 - 支持一次规划多个动作

    核心能力：
    1. 智能规划 - 根据目标制定执行计划
    2. 工具调用 - 动态选择和使用工具
    3. 结果评估 - 判断是否达成目标
    4. 迭代优化 - 根据反馈调整策略
    """

    @abstractmethod
    def plan(
        self,
        intermediate_steps: List[Tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> Union[List[AgentAction], AgentFinish]:
        """
        规划下一步动作

        Args:
            intermediate_steps: 已执行的步骤和结果
            callbacks: 回调处理器
            **kwargs: 额外参数

        Returns:
            AgentAction列表或AgentFinish
        """
        pass

    @abstractmethod
    async def aplan(
        self,
        intermediate_steps: List[Tuple[AgentAction, str]],
        callbacks: Callbacks = None,
        **kwargs: Any,
    ) -> Union[List[AgentAction], AgentFinish]:
        """异步规划"""
        pass

class AgentExecutor(Chain):
    """
    Agent执行器 - 管理Agent的完整执行流程

    执行模式：
    1. 迭代执行 - 循环规划和执行直到完成
    2. 工具管理 - 维护可用工具集合
    3. 异常处理 - 处理执行过程中的各种异常
    4. 结果收集 - 收集和整理执行结果
    """

    agent: Union[BaseAgent, Runnable]
    tools: Sequence[BaseTool]
    return_intermediate_steps: bool = False
    max_iterations: Optional[int] = 15
    max_execution_time: Optional[float] = None
    early_stopping_method: str = "force"
    handle_parsing_errors: Union[bool, str, Callable[[OutputParserException], str]] = False

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Dict[str, Any]:
        """
        Agent执行的核心流程

        执行步骤：
        1. 初始化执行环境
        2. 开始迭代执行循环
        3. Agent规划下一步动作
        4. 执行工具调用
        5. 收集执行结果
        6. 判断是否继续或结束
        """
        # 设置工具名称映射
        name_to_tool_map = {tool.name: tool for tool in self.tools}

        # 颜色映射（用于日志显示）
        color_mapping = get_color_mapping(
            [tool.name for tool in self.tools], excluded_colors=["green", "red"]
        )

        # 初始化执行状态
        intermediate_steps: List[Tuple[AgentAction, str]] = []
        iterations = 0
        time_elapsed = 0.0
        start_time = time.time()

        # 执行循环
        while self._should_continue(iterations, time_elapsed):
            # 1. Agent规划
            next_step_output = self._take_next_step(
                name_to_tool_map,
                color_mapping,
                inputs,
                intermediate_steps,
                run_manager=run_manager,
            )

            # 2. 检查是否完成
            if isinstance(next_step_output, AgentFinish):
                return self._return(
                    next_step_output, intermediate_steps, run_manager=run_manager
                )

            # 3. 更新执行状态
            intermediate_steps.extend(next_step_output)
            if len(next_step_output) == 1:
                next_step_action = next_step_output[0]
                # 记录工具使用
                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
                if next_step_action[0].tool in tool_run_kwargs:
                    make_tools_runnable_with_chat_history(
                        [name_to_tool_map[next_step_action[0].tool]], **tool_run_kwargs
                    )

            iterations += 1
            time_elapsed = time.time() - start_time

        # 达到最大迭代次数或时间限制
        output = self.agent.return_stopped_response(
            self.early_stopping_method, intermediate_steps, **inputs
        )
        return self._return(output, intermediate_steps, run_manager=run_manager)

    def _take_next_step(
        self,
        name_to_tool_map: Dict[str, BaseTool],
        color_mapping: Dict[str, str],
        inputs: Dict[str, Any],
        intermediate_steps: List[Tuple[AgentAction, str]],
        run_manager: Optional[CallbackManagerForChainRun] = None,
    ) -> Union[AgentFinish, List[Tuple[AgentAction, str]]]:
        """执行下一步动作"""
        try:
            # 1. Agent规划
            intermediate_steps = self._prepare_intermediate_steps(intermediate_steps)

            # 获取Agent的输出
            output = self.agent.plan(
                intermediate_steps,
                callbacks=run_manager.get_child() if run_manager else None,
                **inputs,
            )
        except OutputParserException as e:
            # 处理解析错误
            if isinstance(self.handle_parsing_errors, bool):
                raise_error = not self.handle_parsing_errors
            else:
                raise_error = False

            if raise_error:
                raise ValueError(
                    "An output parsing error occurred. "
                    "In order to pass this error back to the agent and have it try "
                    "again, pass `handle_parsing_errors=True` to the AgentExecutor. "
                    f"This is the error: {str(e)}"
                )

            # 使用错误处理策略
            if isinstance(self.handle_parsing_errors, str):
                observation = self.handle_parsing_errors
            elif callable(self.handle_parsing_errors):
                observation = self.handle_parsing_errors(e)
            else:
                observation = "Invalid or incomplete response"

            output = AgentAction("_Exception", observation, "")

        # 2. 处理Agent输出
        if isinstance(output, AgentFinish):
            return output

        actions: List[AgentAction]
        if isinstance(output, AgentAction):
            actions = [output]
        else:
            actions = output

        # 3. 执行动作
        result = []
        for agent_action in actions:
            if run_manager:
                run_manager.on_agent_action(agent_action, color="green")

            # 查找并执行工具
            if agent_action.tool in name_to_tool_map:
                tool = name_to_tool_map[agent_action.tool]
                return_direct = tool.return_direct
                color = color_mapping[agent_action.tool]

                # 执行工具
                tool_run_kwargs = self.agent.tool_run_logging_kwargs()
                if agent_action.tool in tool_run_kwargs:
                    tool_run_kwargs = tool_run_kwargs[agent_action.tool]
                else:
                    tool_run_kwargs = {}

                if return_direct:
                    tool_run_kwargs["llm_prefix"] = ""

                # 实际执行工具
                observation = tool.run(
                    agent_action.tool_input,
                    verbose=self.verbose,
                    color=color,
                    callbacks=run_manager.get_child() if run_manager else None,
                    **tool_run_kwargs,
                )
            else:
                # 工具不存在的处理
                observation = f"Tool {agent_action.tool} not found."
                if run_manager:
                    run_manager.on_tool_error(
                        ValueError(observation), color="red"
                    )

            result.append((agent_action, observation))

        return result

# 单机部署
services:
  langchain-app:
    image: langchain-app:latest
    ports: ["8000:8000"]
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on: [redis]

  redis:
    image: redis:7-alpine
    volumes: [redis_data:/data]

# 微服务架构
services:
  api-gateway:
    image: nginx:alpine
    ports: ["80:80", "443:443"]

  langchain-api:
    image: langchain-api:latest
    replicas: 3
    environment:
      - REDIS_CLUSTER=redis-cluster:6379
      - POSTGRES_URL=postgres://db:5432/langchain

  langchain-worker:
    image: langchain-worker:latest
    replicas: 5
    environment:
      - CELERY_BROKER=redis://redis:6379

# Kubernetes集群部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: langchain-app
spec:
  replicas: 10
  template:
    spec:
      containers:
      - name: app
        image: langchain-app:latest
        resources:
          requests: {memory: "512Mi", cpu: "250m"}
          limits: {memory: "1Gi", cpu: "500m"}

---
apiVersion: v1
kind: Service
metadata:
  name: langchain-service
spec:
  type: LoadBalancer
  ports: [{port: 80, targetPort: 8000}]

# Prometheus告警规则
groups:
- name: langchain.rules
  rules:
  - alert: HighErrorRate
    expr: rate(langchain_requests_total{status="error"}[5m]) > 0.1
    for: 2m
    annotations:
      summary: "LangChain error rate is high"

  - alert: HighLatency
    expr: histogram_quantile(0.95, rate(langchain_request_duration_seconds_bucket[5m])) > 2
    for: 5m
    annotations:
      summary: "LangChain 95th percentile latency is high"