LangChain 设计与实现

第16章 序列化与配置系统

作者 杨艺韬 · 7,977 字

第16章 序列化与配置系统

开篇引言

在 LangChain 的实际应用中,我们经常需要将构建好的链、模型和提示模板保存下来,以便在不同环境中复用,或者在运行时根据需求动态切换配置。这两个需求看似简单,实则牵涉到一系列深层次的设计问题:如何安全地序列化包含 API 密钥的对象?如何在反序列化时防止恶意代码注入?如何在不重建对象的情况下切换底层模型?

LangChain 的解决方案分为两个互补的子系统:序列化系统langchain_core.load)负责对象的持久化和恢复,配置系统ConfigurableField / RunnableConfig)负责运行时的动态参数调整。两者共同构成了 LangChain 的"状态管理"基础设施。

本章将深入这两个系统的源码实现,从 Serializable 基类的设计到 Reviver 的安全模型,从 ConfigurableField 的声明式配置到 DynamicRunnable 的延迟绑定机制。

本章要点

  • Serializable 基类的设计:lc_id、lc_secrets、lc_attributes 的作用
  • dumps/dumpd 的序列化流程与注入防护(escape 机制)
  • loads/load 的反序列化流程与白名单安全模型
  • Reviver 类的多层安全防护:命名空间验证、类路径白名单、init_validator
  • ConfigurableField/ConfigurableFieldSpec 的声明式配置
  • DynamicRunnable 的延迟绑定机制
  • RunnableConfig 的结构与传播

16.1 Serializable 基类

序列化系统的根基是 Serializable 类,定义在 langchain_core/load/serializable.py 中。它继承自 Pydantic 的 BaseModel 和 Python 的 ABC

16.1.1 类设计

# langchain_core/load/serializable.py

class Serializable(BaseModel, ABC):
    """序列化基类"""

    @classmethod
    def is_lc_serializable(cls) -> bool:
        """此类是否可序列化?默认 False"""
        return False

    @classmethod
    def get_lc_namespace(cls) -> list[str]:
        """获取命名空间,用于序列化标识符"""
        return cls.__module__.split(".")

    @property
    def lc_secrets(self) -> dict[str, str]:
        """构造参数名到密钥 ID 的映射"""
        return {}

    @property
    def lc_attributes(self) -> dict:
        """额外需要序列化的属性"""
        return {}

    @classmethod
    def lc_id(cls) -> list[str]:
        """返回唯一标识符"""
        return [*cls.get_lc_namespace(), cls.__name__]

    model_config = ConfigDict(extra="ignore")

这段代码中包含了几个关键的设计决策,每一个都值得深入分析。

第一个决策是默认不可序列化。is_lc_serializable 方法默认返回 False。即使一个类继承了 Serializable,它也不会自动获得序列化能力,必须显式地覆盖这个方法并返回 True。这是一种"安全默认"策略。序列化意味着类的内部状态(包括可能包含的敏感信息)会被暴露为明文数据,反序列化意味着类可以被远程实例化。只有开发者明确意识到这些安全影响并同意承担时,才应该启用序列化。如果反过来,默认允许序列化,那么开发者可能在不知情的情况下暴露了不该暴露的信息。

第二个决策是使用命名空间作为身份标识。get_lc_namespace 从模块路径自动生成命名空间。例如 langchain_openai.chat_models.base.ChatOpenAI 的命名空间就是 ["langchain_openai", "chat_models", "base"],再加上类名组成完整的 lc_id。这种基于模块路径的自动生成机制避免了手动维护标识符的负担,但也意味着如果类在模块之间移动,它的标识符会改变。LangChain 通过 SERIALIZABLE_MAPPING 映射表来处理这种情况,在旧标识符和新标识符之间建立桥接。

第三个决策是显式的密钥保护机制。lc_secrets 属性声明了哪些构造参数包含敏感信息,以及它们对应的环境变量名。序列化时,这些字段的值会被替换为 SerializedSecret 标记,标记中只包含环境变量名而非实际值。这种设计确保了序列化后的数据可以安全地存储和传输,而密钥的实际值只在反序列化时通过 secrets_map 或环境变量恢复。

第四个决策是 lc_attributes 的存在。有些重要的状态信息不是构造参数,而是在初始化后计算或设置的属性。lc_attributes 允许将这些属性也纳入序列化范围,但前提是它们必须可以通过构造函数重新设置。这种限制确保了反序列化的一致性 -- 所有状态都通过构造函数重建,不存在"额外初始化"的步骤。

16.1.2 to_json -- 序列化核心

to_json 方法将对象转换为可序列化的字典:

def to_json(self) -> SerializedConstructor | SerializedNotImplemented:
    if not self.is_lc_serializable():
        return self.to_json_not_implemented()

    model_fields = type(self).model_fields
    secrets = {}
    lc_kwargs = {}

    # 收集所有有用的字段值
    for k, v in self:
        if not _is_field_useful(self, k, v):
            continue
        if k in model_fields and model_fields[k].exclude:
            continue
        lc_kwargs[k] = getattr(self, k, v)

    # 从 MRO 链中合并 lc_secrets 和 lc_attributes
    for cls in [None, *self.__class__.mro()]:
        if cls is Serializable:
            break
        this = cast(Serializable, self if cls is None else super(cls, self))
        secrets.update(this.lc_secrets)
        # 处理别名
        for key in list(secrets):
            value = secrets[key]
            if (key in model_fields) and (
                alias := model_fields[key].alias
            ) is not None:
                secrets[alias] = value
        lc_kwargs.update(this.lc_attributes)

    # 确保所有密钥字段都被包含
    for key in secrets:
        secret_value = getattr(self, key, None) or lc_kwargs.get(key)
        if secret_value is not None:
            lc_kwargs.update({key: secret_value})

    return {
        "lc": 1,
        "type": "constructor",
        "id": self.lc_id(),
        "kwargs": lc_kwargs if not secrets
                  else _replace_secrets(lc_kwargs, secrets),
    }

序列化输出的结构是一个固定格式的字典:

{
    "lc": 1,                              # 序列化格式版本
    "type": "constructor",                 # 类型标识
    "id": ["langchain_openai", "chat_models", "base", "ChatOpenAI"],
    "kwargs": {
        "model_name": "gpt-4",
        "temperature": 0.7,
        "openai_api_key": {               # 密钥被替换为标记
            "lc": 1,
            "type": "secret",
            "id": ["OPENAI_API_KEY"]
        }
    }
}
flowchart TD
    A["obj.to_json()"] --> B{is_lc_serializable?}
    B -->|否| C["to_json_not_implemented()"]
    B -->|是| D["遍历模型字段<br/>收集 lc_kwargs"]
    D --> E["遍历 MRO<br/>合并 lc_secrets + lc_attributes"]
    E --> F{"有密钥字段?"}
    F -->|是| G["_replace_secrets<br/>将密钥值替换为 secret 标记"]
    F -->|否| H["直接使用 lc_kwargs"]
    G --> I["返回 SerializedConstructor<br/>lc=1, type='constructor'<br/>id=lc_id(), kwargs=..."]
    H --> I
    C --> J["返回 SerializedNotImplemented<br/>lc=1, type='not_implemented'"]

16.1.3 _is_field_useful -- 智能字段过滤

并非所有字段都需要序列化。_is_field_useful 函数实现了智能过滤:

def _is_field_useful(inst: Serializable, key: str, value: Any) -> bool:
    field = type(inst).model_fields.get(key)
    if not field:
        return False
    if field.is_required():
        return True              # 必填字段始终包含

    try:
        value_is_truthy = bool(value)
    except Exception:
        value_is_truthy = False

    if value_is_truthy:
        return True              # 非空值包含

    # 空列表/空字典如果是默认值,跳过
    if field.default_factory is dict and isinstance(value, dict):
        return False
    if field.default_factory is list and isinstance(value, list):
        return False

    # 与默认值不同的 falsy 值也包含(如 0、False)
    return _try_neq_default(value, field)

这种过滤策略确保序列化输出尽可能紧凑:默认值的字段不包含在内,但非默认的 falsy 值(如 temperature=0)会被正确保留。

过滤逻辑的复杂性反映了序列化场景下的多种边界情况。必填字段始终保留,因为反序列化时它们是构造函数所必需的。有值的可选字段保留,因为它们可能被用户有意设置为非默认值。空列表和空字典如果是默认值则跳过,因为它们可以在构造时自动创建。最精妙的是对 falsy 但非默认值的处理 -- 比如当用户将温度设为零时,虽然 bool(0)False,但零不等于默认值 0.7,因此应该被保留。如果忽略了这种情况,反序列化后的对象就会使用默认温度而非用户指定的零温度,导致行为不一致。

特别值得一提的是对 Pandas DataFrame 等特殊对象的容错处理。这些对象的布尔求值和相等性比较可能抛出异常或返回非布尔值。代码中的多重 try-except 确保了即使遇到这种异常的对象类型,过滤逻辑也不会崩溃。这种防御性编程风格在序列化这种"基础设施级"代码中尤为重要,因为它需要处理任意用户定义的数据类型。

16.2 dumps 与 dumpd -- 序列化 API

dumpsdumpd 是面向用户的序列化 API。

16.2.1 dumpd -- 转字典

# langchain_core/load/dump.py

def dumpd(obj: Any) -> Any:
    """将对象转换为可 JSON 序列化的字典"""
    obj = _dump_pydantic_models(obj)  # 处理嵌套 Pydantic 模型
    return _serialize_value(obj)

16.2.2 dumps -- 转 JSON 字符串

def dumps(obj: Any, *, pretty: bool = False, **kwargs: Any) -> str:
    """将对象转换为 JSON 字符串"""
    if "default" in kwargs:
        raise ValueError("`default` should not be passed to dumps")

    obj = _dump_pydantic_models(obj)
    serialized = _serialize_value(obj)

    if pretty:
        indent = kwargs.pop("indent", 2)
        return json.dumps(serialized, indent=indent, **kwargs)
    return json.dumps(serialized, **kwargs)

16.2.3 注入防护 -- 转义机制

_serialize_value 是序列化的核心递归函数,它实现了关键的注入防护:

# langchain_core/load/_validation.py

_LC_ESCAPED_KEY = "__lc_escaped__"

def _needs_escaping(obj: dict[str, Any]) -> bool:
    """检查字典是否需要转义"""
    return "lc" in obj or (len(obj) == 1 and _LC_ESCAPED_KEY in obj)

def _serialize_value(obj: Any) -> Any:
    if isinstance(obj, Serializable):
        return _serialize_lc_object(obj)  # LC 对象正常序列化
    if isinstance(obj, dict):
        if _needs_escaping(obj):
            return {_LC_ESCAPED_KEY: obj}  # 危险字典被转义
        return {k: _serialize_value(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_serialize_value(item) for item in obj]
    if isinstance(obj, (str, int, float, bool, type(None))):
        return obj
    return to_json_not_implemented(obj)

这段代码的安全逻辑是:当一个普通字典恰好包含 "lc" 键时(这可能是用户数据碰巧包含了与 LC 序列化格式相同的结构),它会被包装为 {"__lc_escaped__": {...}}。反序列化时,遇到这种包装会直接还原为普通字典,而不会被误认为是 LC 对象而被实例化。

flowchart TD
    A["_serialize_value(obj)"] --> B{obj 类型}
    B -->|Serializable| C["_serialize_lc_object(obj)<br/>正常 LC 序列化"]
    B -->|dict| D{包含 'lc' 键?}
    D -->|是| E["转义: {'__lc_escaped__': obj}<br/>防止被误解为 LC 对象"]
    D -->|否| F["递归: {k: _serialize_value(v)}"]
    B -->|list/tuple| G["递归: [_serialize_value(item)]"]
    B -->|基本类型| H["原样返回"]
    B -->|其他| I["to_json_not_implemented(obj)"]

16.3 loads 与 load -- 反序列化 API

反序列化是安全敏感的操作:它需要根据序列化数据实例化 Python 对象,执行构造函数。如果不加控制,恶意数据可能导致任意代码执行。

16.3.1 loads 和 load 函数

# langchain_core/load/load.py

@beta()
def loads(
    text: str,
    *,
    allowed_objects: Iterable[AllowedObject] | Literal["all", "core"] = "core",
    secrets_map: dict[str, str] | None = None,
    valid_namespaces: list[str] | None = None,
    secrets_from_env: bool = False,
    additional_import_mappings: dict | None = None,
    ignore_unserializable_fields: bool = False,
    init_validator: InitValidator | None = default_init_validator,
) -> Any:
    raw_obj = json.loads(text)
    return load(raw_obj, ...)

@beta()
def load(
    obj: Any,
    *,
    allowed_objects = "core",
    secrets_map = None,
    ...
) -> Any:
    reviver = Reviver(
        allowed_objects, secrets_map, valid_namespaces,
        secrets_from_env, additional_import_mappings,
        ignore_unserializable_fields=ignore_unserializable_fields,
        init_validator=init_validator,
    )

    def _load(obj: Any) -> Any:
        if isinstance(obj, dict):
            # 首先检查是否是转义字典
            if _is_escaped_dict(obj):
                return _unescape_value(obj)  # 还原为普通字典
            # 递归处理子元素,然后应用 Reviver
            loaded_obj = {k: _load(v) for k, v in obj.items()}
            return reviver(loaded_obj)
        if isinstance(obj, list):
            return [_load(o) for o in obj]
        return obj

    return _load(obj)

load 函数的处理流程分为三步:

  1. 检查转义字典并还原
  2. 递归处理嵌套结构
  3. 通过 Reviver 将 LC 对象字典实例化为 Python 对象

16.3.2 Reviver -- 安全的对象恢复

Reviver 是反序列化的核心,它实现了多层安全防护:

class Reviver:
    def __init__(
        self,
        allowed_objects: Iterable[AllowedObject] | Literal["all", "core"] = "core",
        secrets_map: dict[str, str] | None = None,
        valid_namespaces: list[str] | None = None,
        secrets_from_env: bool = False,
        additional_import_mappings: dict | None = None,
        *,
        ignore_unserializable_fields: bool = False,
        init_validator: InitValidator | None = default_init_validator,
    ) -> None:
        self.secrets_from_env = secrets_from_env
        self.secrets_map = secrets_map or {}
        # 默认可信命名空间
        self.valid_namespaces = (
            [*DEFAULT_NAMESPACES, *valid_namespaces]
            if valid_namespaces else DEFAULT_NAMESPACES
        )
        # 计算允许的类路径
        if allowed_objects in ("all", "core"):
            self.allowed_class_paths = (
                _get_default_allowed_class_paths(allowed_objects).copy()
            )
        else:
            self.allowed_class_paths = _compute_allowed_class_paths(
                allowed_objects, self.import_mappings
            )
        self.init_validator = init_validator

默认的可信命名空间包括:

DEFAULT_NAMESPACES = [
    "langchain",
    "langchain_core",
    "langchain_community",
    "langchain_anthropic",
    "langchain_groq",
    "langchain_google_genai",
    "langchain_aws",
    "langchain_openai",
    "langchain_google_vertexai",
    "langchain_mistralai",
    "langchain_fireworks",
    "langchain_xai",
    "langchain_sambanova",
    "langchain_perplexity",
]

16.3.3 Reviver.call -- 三阶段安全验证

def __call__(self, value: dict[str, Any]) -> Any:
    # 阶段 1:处理密钥
    if value.get("lc") == 1 and value.get("type") == "secret":
        [key] = value["id"]
        if key in self.secrets_map:
            return self.secrets_map[key]
        if self.secrets_from_env and key in os.environ:
            return os.environ[key]
        return None

    # 阶段 2:处理不可序列化标记
    if value.get("lc") == 1 and value.get("type") == "not_implemented":
        if self.ignore_unserializable_fields:
            return None
        raise NotImplementedError(...)

    # 阶段 3:处理构造函数类型
    if value.get("lc") == 1 and value.get("type") == "constructor":
        [*namespace, name] = value["id"]
        mapping_key = tuple(value["id"])

        # 安全检查 1:白名单验证
        if (self.allowed_class_paths is not None
            and mapping_key not in self.allowed_class_paths):
            raise ValueError(
                f"Deserialization of {mapping_key!r} is not allowed."
            )

        # 安全检查 2:命名空间验证
        if namespace[0] not in self.valid_namespaces:
            raise ValueError(f"Invalid namespace: {value}")

        # 安全检查 3:导入路径验证
        if mapping_key in self.import_mappings:
            import_path = self.import_mappings[mapping_key]
            import_dir, name = import_path[:-1], import_path[-1]
        elif namespace[0] in DISALLOW_LOAD_FROM_PATH:
            raise ValueError(...)
        else:
            import_dir = namespace

        if import_dir[0] not in self.valid_namespaces:
            raise ValueError(f"Invalid namespace: {value}")

        kwargs = value.get("kwargs", {})

        # 安全检查 4:类特定验证器
        if mapping_key in CLASS_INIT_VALIDATORS:
            CLASS_INIT_VALIDATORS[mapping_key](mapping_key, kwargs)

        # 安全检查 5:通用验证器(如阻止 jinja2 模板)
        if self.init_validator is not None:
            self.init_validator(mapping_key, kwargs)

        # 安全检查通过,执行导入和实例化
        mod = importlib.import_module(".".join(import_dir))
        cls = getattr(mod, name)

        # 最终检查:必须是 Serializable 子类
        if not issubclass(cls, Serializable):
            raise ValueError(f"Invalid namespace: {value}")

        return cls(**kwargs)

    return value
flowchart TD
    A["Reviver(value)"] --> B{type == 'secret'?}
    B -->|是| C["从 secrets_map 或环境变量获取密钥值"]
    B -->|否| D{type == 'not_implemented'?}
    D -->|是| E["抛出 NotImplementedError<br/>或返回 None"]
    D -->|否| F{type == 'constructor'?}
    F -->|否| G["原样返回 value"]
    F -->|是| H["安全检查 1: 白名单"]
    H --> I["安全检查 2: 命名空间"]
    I --> J["安全检查 3: 导入路径"]
    J --> K["安全检查 4: 类特定验证器"]
    K --> L["安全检查 5: 通用验证器<br/>(阻止 jinja2 等)"]
    L --> M["importlib.import_module"]
    M --> N{是 Serializable 子类?}
    N -->|是| O["cls(**kwargs)<br/>实例化对象"]
    N -->|否| P["抛出 ValueError"]

16.3.4 allowed_objects 的三种模式

allowed_objects: Iterable[AllowedObject] | Literal["all", "core"] = "core"
模式 说明 安全等级
"core" 仅允许 langchain_core 中的类 最高
"all" 允许所有映射中注册的类(含 Partner 包) 中等
[AIMessage, ...] 仅允许指定的类 自定义

推荐在生产环境中使用显式列表模式,精确控制可反序列化的类型。这种最小权限原则是安全工程的基本实践 -- 只允许确实需要的类型,而非宽泛地信任整个命名空间。

三种模式的选择反映了不同的信任等级。"core" 模式适合处理来自外部的序列化数据(如用户上传的配置),因为 langchain_core 中的类(消息、文档、提示模板等)在初始化时不会执行网络请求或文件操作。"all" 模式适合内部系统之间的数据交换,因为 Partner 包中的类可能在初始化时建立数据库连接或 HTTP 客户端,但这些行为在受信环境中是可接受的。显式列表模式适合安全要求最高的场景,如处理不可信的 webhook 数据。

另一个重要的安全细节是 DISALLOW_LOAD_FROM_PATH 列表。某些命名空间(如 langchain_communitylangchain)只允许通过映射表加载,不允许直接按路径导入。这是因为这些命名空间中可能包含大量未经审核的第三方集成代码,允许按路径导入可能会实例化不安全的类。通过映射表加载则确保了只有明确注册过的类才能被反序列化。

16.3.5 密钥恢复的安全考量

# 反序列化时恢复密钥
secrets_from_env: bool = False  # 默认关闭

secrets_from_env=False 是一个重要的安全默认值。如果设为 True,恶意的序列化数据可以在 secret 字段中指定任意环境变量名,导致在反序列化时泄露敏感信息。只有在完全信任数据来源时才应启用。

16.4 ConfigurableField 与动态配置

LangChain 的配置系统允许在不重建对象的情况下,运行时动态调整参数。

16.4.1 ConfigurableField 家族

# langchain_core/runnables/utils.py

class ConfigurableField(NamedTuple):
    """可配置字段"""
    id: str             # 唯一标识符
    name: str | None = None
    description: str | None = None
    annotation: Any | None = None
    is_shared: bool = False

class ConfigurableFieldSingleOption(NamedTuple):
    """单选可配置字段"""
    id: str
    options: Mapping[str, Any]    # 可选项映射
    default: str                  # 默认选项键
    name: str | None = None
    description: str | None = None
    is_shared: bool = False

class ConfigurableFieldMultiOption(NamedTuple):
    """多选可配置字段"""
    id: str
    options: Mapping[str, Any]
    default: Sequence[str]        # 默认选项键列表
    name: str | None = None
    description: str | None = None
    is_shared: bool = False

class ConfigurableFieldSpec(NamedTuple):
    """可配置字段规范"""
    id: str
    annotation: Any               # 类型注解
    name: str | None = None
    description: str | None = None
    default: Any = None
    is_shared: bool = False
    dependencies: list[str] | None = None

这四种配置字段类型覆盖了不同的使用场景:

16.4.2 configurable_fields -- 声明可配置项

Runnable 的 configurable_fields 方法用于声明哪些字段是可配置的:

# 使用示例
from langchain_core.runnables import ConfigurableField
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 声明 model_name 和 temperature 为可配置字段
configurable_model = model.configurable_fields(
    model_name=ConfigurableField(
        id="model_name",
        name="模型名称",
        description="要使用的 OpenAI 模型",
    ),
    temperature=ConfigurableField(
        id="temperature",
        name="温度",
        description="生成的随机性控制",
    ),
)

# 运行时动态配置
result = configurable_model.invoke(
    "Hello",
    config={"configurable": {"model_name": "gpt-4", "temperature": 0.9}},
)

16.4.3 configurable_alternatives -- 整体替换

configurable_alternatives 允许在运行时切换整个 Runnable 实现:

from langchain_anthropic import ChatAnthropic
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4").configurable_alternatives(
    ConfigurableField(id="llm"),
    default_key="openai",
    anthropic=ChatAnthropic(model="claude-3-sonnet-20240229"),
)

# 使用默认的 OpenAI
result1 = model.invoke("Hello")

# 切换到 Anthropic
result2 = model.invoke(
    "Hello",
    config={"configurable": {"llm": "anthropic"}},
)
flowchart TD
    subgraph "configurable_fields"
        A["ChatOpenAI(model='gpt-3.5')"] -->|".configurable_fields()"| B["DynamicRunnable"]
        B -->|"config: model_name='gpt-4'"| C["ChatOpenAI(model='gpt-4')"]
        B -->|"config: temperature=0.9"| D["ChatOpenAI(temperature=0.9)"]
    end

    subgraph "configurable_alternatives"
        E["ChatOpenAI(默认)"] -->|".configurable_alternatives()"| F["DynamicRunnable"]
        F -->|"config: llm='openai'"| G["ChatOpenAI"]
        F -->|"config: llm='anthropic'"| H["ChatAnthropic"]
    end

16.4.4 DynamicRunnable -- 延迟绑定

DynamicRunnable 是配置系统的运行时载体:

# langchain_core/runnables/configurable.py

class DynamicRunnable(RunnableSerializable[Input, Output]):
    default: RunnableSerializable[Input, Output]
    config: RunnableConfig | None = None

    def prepare(
        self, config: RunnableConfig | None = None
    ) -> tuple[Runnable[Input, Output], RunnableConfig]:
        """根据配置准备实际的 Runnable"""
        runnable: Runnable[Input, Output] = self
        while isinstance(runnable, DynamicRunnable):
            runnable, config = runnable._prepare(
                merge_configs(runnable.config, config)
            )
        return runnable, cast(RunnableConfig, config)

    def invoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs
    ) -> Output:
        runnable, config = self.prepare(config)
        return runnable.invoke(input, config, **kwargs)

    async def ainvoke(
        self, input: Input, config: RunnableConfig | None = None, **kwargs
    ) -> Output:
        runnable, config = self.prepare(config)
        return await runnable.ainvoke(input, config, **kwargs)

prepare 方法是关键。每次调用 invoke 时,它根据传入的配置动态解析出实际应该使用的 Runnable。如果有嵌套的 DynamicRunnable(多层配置),它会循环解包直到获得最终的具体 Runnable。

这种"延迟绑定"设计意味着 DynamicRunnable 本身不执行任何逻辑,它只是一个"配置分发器"。真正的执行总是委托给解析出的具体 Runnable。

理解这个设计需要区分"构建时"和"运行时"两个阶段。在构建时(调用 configurable_fieldsconfigurable_alternatives 时),系统创建一个 DynamicRunnable 作为占位符,记录可配置的字段和备选方案。在运行时(调用 invoke 时),系统根据实际传入的 RunnableConfig 解析出应该使用的具体 Runnable 实例,然后将调用委托给它。

这种两阶段设计带来了一个重要的好处:线程安全。DynamicRunnable 不持有任何可变状态,prepare 方法每次都根据传入的 config 参数创建新的实例。多个线程可以同时对同一个 DynamicRunnable 调用 invoke,传入不同的配置,而不会产生竞态条件。这在 Web 服务器环境中非常重要,因为不同的请求可能需要使用不同的模型配置,但它们共享同一个 DynamicRunnable 实例。

另一个微妙的设计是 while isinstance(runnable, DynamicRunnable) 循环。这意味着 DynamicRunnable 可以嵌套 -- 一个可配置的 Runnable 的某个备选方案本身也可以是可配置的。循环解包确保了无论嵌套多深,最终都能获得一个具体的可执行 Runnable。这种递归解包的设计使得配置系统具有了无限的组合能力。

16.5 RunnableConfig -- 运行时配置传播

RunnableConfig 是贯穿整个 Runnable 调用链的配置容器:

# langchain_core/runnables/config.py

class RunnableConfig(TypedDict, total=False):
    tags: list[str]
    """标签,用于过滤和追踪"""

    metadata: dict[str, Any]
    """元数据,传递给回调"""

    callbacks: Callbacks
    """回调处理器"""

    run_name: str
    """运行名称,用于追踪"""

    max_concurrency: int | None
    """最大并发数"""

    run_id: uuid.UUID
    """运行唯一标识"""

    configurable: dict[str, Any]
    """可配置参数,用于 ConfigurableField"""

total=False 的 TypedDict 设计允许部分配置,通过 merge_configs 进行合并:

# 配置合并:子配置继承父配置
parent_config = {"tags": ["production"], "metadata": {"user": "alice"}}
child_config = {"tags": ["chain-step-1"]}

merged = merge_configs(parent_config, child_config)
# 结果: {"tags": ["production", "chain-step-1"], "metadata": {"user": "alice"}}

configurable 字段是配置系统的入口。当用户传入 config={"configurable": {"model_name": "gpt-4"}} 时,DynamicRunnable.prepare() 从这个字段中读取配置值,创建相应的 Runnable 实例。

flowchart TB
    subgraph "RunnableConfig 结构"
        A["tags: ['prod']"]
        B["metadata: {'user': 'alice'}"]
        C["callbacks: [handler]"]
        D["run_name: 'my-chain'"]
        E["max_concurrency: 5"]
        F["configurable: {'model': 'gpt-4'}"]
    end

    subgraph "传播路径"
        G["chain.invoke(input, config)"] --> H["chain 读取 config"]
        H --> I["child.invoke(input, child_config)"]
        I --> J["child 继承并合并 config"]
    end

    A --> G
    B --> G
    C --> G
    D --> G
    E --> G
    F --> G

16.6 安全模型深度分析

LangChain 的序列化安全模型是多层防御的:

第一层:转义防护

序列化时,包含 "lc" 键的普通字典被自动转义为 {"__lc_escaped__": ...}。反序列化时,转义字典被还原为普通字典,绝不会被实例化为对象。这确保了用户数据(如 metadata)中碰巧包含 "lc" 键的情况不会被误解。

第二层:白名单控制

allowed_objects 参数控制哪些类可以被反序列化。默认的 "core" 模式只允许 langchain_core 中的类,是最严格的策略。即使恶意数据包含完整的类路径和构造参数,如果类不在白名单中,反序列化就会被拒绝。

第三层:命名空间验证

即使类在白名单中,其命名空间也必须属于可信列表。这防止了通过篡改类路径指向恶意模块的攻击。

第四层:init_validator

反序列化前会调用验证器检查构造参数。默认验证器阻止 template_format="jinja2",防止 Jinja2 模板注入攻击(Jinja2 模板可以执行任意 Python 代码)。

第五层:Serializable 子类检查

最终的安全网:导入的类必须是 Serializable 的子类。这确保只有明确声明为可序列化的类才能被实例化。

flowchart TB
    A["反序列化数据"] --> B{"转义字典?<br/>__lc_escaped__"}
    B -->|是| C["还原为普通字典<br/>(不实例化)"]
    B -->|否| D{"类路径在白名单?<br/>allowed_class_paths"}
    D -->|否| E["拒绝: ValueError"]
    D -->|是| F{"命名空间可信?<br/>valid_namespaces"}
    F -->|否| E
    F -->|是| G{"CLASS_INIT_VALIDATORS<br/>类特定验证"}
    G -->|失败| E
    G -->|通过| H{"init_validator<br/>通用验证 (如 jinja2 阻止)"}
    H -->|失败| E
    H -->|通过| I["importlib.import_module"]
    I --> J{"issubclass(cls, Serializable)?"}
    J -->|否| E
    J -->|是| K["cls(**kwargs)<br/>安全实例化"]

16.7 设计决策分析

为什么序列化默认关闭?

is_lc_serializable() 默认返回 False,这是一个重要的安全决策。序列化意味着类的构造参数会被暴露,反序列化意味着类可以被远程实例化。只有开发者明确意识到并同意这些影响时,才应该启用序列化。

TypedDict vs Pydantic 用于 RunnableConfig

RunnableConfig 使用 TypedDict 而非 Pydantic 模型,原因有二:首先,config 在每次 Runnable 调用时都会被创建和传播,TypedDict 比 Pydantic 模型轻量得多;其次,total=False 的 TypedDict 天然支持部分配置,不需要所有字段都有值。

映射表 vs 反射 用于类路径解析

LangChain 使用 SERIALIZABLE_MAPPING 映射表来解析类路径,而非纯粹依赖 Python 的反射机制。这使得类可以在包之间迁移(例如从 langchain 迁移到 langchain_openai),旧的序列化数据仍然可以被正确反序列化。映射表也充当了安全白名单的角色。

配置系统的"不可变"语义

DynamicRunnable 不修改原始 Runnable,而是在每次调用时创建新的配置实例。这种"写时复制"语义确保了线程安全:多个并发调用可以使用不同的配置而互不干扰。

16.8 序列化格式深度分析

LangChain 的序列化格式是一种自描述的 JSON 结构,其设计经过了多轮迭代。让我们深入分析这种格式的设计考量。

序列化输出总是一个包含四个固定键的字典:lc(版本号)、type(类型标识)、id(类路径)和 kwargs(构造参数)。lc: 1 是当前的格式版本,预留了未来格式升级的空间。如果格式需要不兼容的变更,版本号可以递增为 2,反序列化器可以根据版本号选择不同的处理逻辑。

type 字段只有三种合法值:"constructor"(可以被重建的对象)、"secret"(密钥标记)和 "not_implemented"(无法序列化的对象)。这三种类型覆盖了所有可能的序列化需求。constructor 类型包含完整的重建信息,反序列化器可以据此实例化对象。secret 类型是一个占位符,反序列化时需要从外部来源获取实际值。not_implemented 类型是一种优雅的降级 -- 对于确实无法序列化的对象(如匿名函数、文件句柄),记录其文本表示以供调试,但明确标记为不可恢复。

id 字段是一个字符串列表,表示类的完整路径。例如 ["langchain_openai", "chat_models", "base", "ChatOpenAI"]。使用列表而非点分字符串的原因是避免歧义 -- 如果类名中包含点号(虽然不常见但在 Python 中是合法的),点分字符串会产生歧义。

kwargs 字段包含了重建对象所需的所有构造参数。这些参数是递归序列化的,意味着嵌套的 Serializable 对象会被递归地转换为相同的 JSON 结构。这使得整个序列化输出是一棵自包含的树,任何节点都可以独立反序列化。

这种设计使得序列化输出具有很好的可读性和可调试性。开发者可以用肉眼阅读 JSON,理解对象的类型和配置。这在调试序列化问题时非常有价值,而二进制序列化格式(如 pickle)就做不到这一点。

16.9 序列化映射表:跨版��兼容

序列化系统中一个容易被忽视但极其重要的组件是 SERIALIZABLE_MAPPING(定义在 langchain_core/load/mapping.py 中)。这张映射表记录了从旧类路径到新类路径的对应关系,使得在包之间迁移类时,旧的序列化数据仍然可以被正确反序列化。

例如,当 ChatOpenAIlangchain.chat_models.openai 迁移到 langchain_openai.chat_models.base 时,映射表中会保留一条记录,将旧路径指向新路径。Reviver 在解析类路径时,先查映射表,找到实际的导入路径后再执行导入。

这张映射表也充当了白名单的角色。_get_default_allowed_class_paths 函数从映射表中提取所有已知的类路径,作为默认的允许列表。这意味着只有在映射表中注册过的类才能被反序列化,新增的类必须先注册才能支持序列化恢复。

对于自定义类,可以通过 additional_import_mappings 参数向 load 函数注入额外的映射,而不需要修改全局映射表。这种设计保持了核心映射表的稳定性,同时允许用户扩展。

16.9 配置系统的实际应用场景

配置系统在实际开发中有几个典型的应用场景,值得深入讨论。

A/B 测试

通过 configurable_alternatives,你可以在不修改代码的情况下切换底层模型,实现 A/B 测试:

model = ChatOpenAI(model="gpt-4").configurable_alternatives(
    ConfigurableField(id="model_provider"),
    default_key="openai",
    anthropic=ChatAnthropic(model="claude-3-sonnet-20240229"),
    groq=ChatGroq(model="llama3-70b-8192"),
)

# 根据用户分组选择不同的模型
config = {"configurable": {"model_provider": user_group}}
result = chain.invoke(input, config=config)

这种方式比硬编码的 if-else 分支更加清晰,而且配置的传播通过 RunnableConfig 自动完成,整条链中所有用到该模型的地方都会同步切换。

多租户温度控制

在多租户场景中,不同客户可能有不同的参数需求。通过 configurable_fields,你可以让同一个模型实例为不同客户提供不同的温度、最大 token 数等参数:

model = ChatOpenAI(model="gpt-4", temperature=0.7).configurable_fields(
    temperature=ConfigurableField(id="temperature"),
    max_tokens=ConfigurableField(id="max_tokens"),
)

# 为创意写作客户设高温度
creative_config = {"configurable": {"temperature": 0.9, "max_tokens": 2000}}
# 为数据分析客户设低温度
analytical_config = {"configurable": {"temperature": 0.1, "max_tokens": 500}}

开发/生产环境切换

在开发环境使用便宜的小模型快速迭代,在生产环境切换到高质量的大模型:

model = ChatOpenAI(model="gpt-3.5-turbo").configurable_fields(
    model_name=ConfigurableField(id="model"),
)

dev_config = {"configurable": {"model": "gpt-3.5-turbo"}}
prod_config = {"configurable": {"model": "gpt-4"}}

配置可以从环境变量、配置文件或请求参数中读取,与代码逻辑完全解耦。

16.10 序列化系统与 LangSmith 的关系

LangChain 的序列化系统与 LangSmith(追踪和监控平台)之间存在紧密的联系。当一个 Runnable 被执行时,其序列化表示会被作为元数据发送到 LangSmith,使得在追踪界面中可以看到每个节点的完整配置。

这也是为什么 to_json 方法要处理密钥替换 -- 密钥值不能出现在追踪数据中。SerializedSecret 类型({"lc": 1, "type": "secret", "id": ["OPENAI_API_KEY"]})确保了只有密钥的名称而非实际值被记录。

同时,to_json_not_implemented 用于处理不可序列化的对象(如自定义函数、lambda 表达式)。这些对象在序列化表示中被标记为 not_implemented,附带 repr 字符串供人工阅读,但不能被反序列化恢复。在追踪场景下,这种退化处理是可接受的 -- 开发者仍然能在 LangSmith 中看到对象的文字描述。

小结

本章深入剖析了 LangChain 的序列化与配置两大子系统。

序列化系统以 Serializable 基类为根,通过 to_json 生成标准化的序列化表示,通过 Reviver 实现安全的反序列化。五层安全防护(转义、白名单、命名空间、init_validator、子类检查)构成了一套纵深防御体系。映射表机制确保了类在跨包迁移后旧数据仍可恢复,体现了对向后兼容性的重视。

配置系统以 ConfigurableField 家族为声明式接口,通过 DynamicRunnable 实现运行时的延迟绑定。RunnableConfig 贯穿整个调用链,将配置参数从顶层传播到每一个子 Runnable。在 A/B 测试、多租户参数控制、环境切换等场景下,配置系统让同一套代码能够服务于不同的需求,而无需条件分支或代码重构。

两个系统共同支撑了 LangChain 应用的"可移植性":序列化使对象可以跨环境传输和持久化,配置使行为可以在运行时动态调整。这种将"对象状态"和"运行时行为"清晰分离的设计,是构建灵活 AI 应用框架的重要基石。下一章,我们将转向 LangChain 的生态层面,看看 Partner 集成架构如何将第三方服务标准化地接入 LangChain 体系。


延伸阅读:序列化的五层安全防护

LangChain 的序列化安全设计——"转义、白名单、命名空间、init_validator、子类检查"五层防护——是纵深防御(defense in depth)思想的典范每一层独立工作、都能拦截某一类攻击;多层叠加、即使某层被绕过、其他层还能兜底这种"不依赖任何单一防线"的设计、是安全工程的基本功

类比建筑安全——一栋商业大楼不会只靠"大门锁好"来防盗、而是"门禁 + 保安 + 摄像头 + 报警器 + 警报联动"多层防护;任何单一防护可能被绕过、但多层叠加后、恶意入侵者几乎不可能全突破软件安全同理——不要相信"一个防线就能解决所有问题"——反序列化安全、XSS 防护、SQL 注入防护、CSRF 防护——每一项都需要多层设计《LangGraph 源码》第 8 章的 checkpoint 安全、《OpenClaw 源码》第 13 章的 Agent 安全——都体现了类似的"多层防御"思想

延伸阅读:向后兼容性的映射表艺术

LangChain 通过"映射表机制"支持"老版本序列化的对象、在新版本反序列化"——即使类从 langchain_core.xxx 迁移到 langchain_openai.xxx、映射表会告诉反序列化逻辑"这两个其实是一个类"这种"维护旧名字到新名字的映射"、是软件演化时保持向后兼容的经典技巧

类似的映射表、在其他系统也有——Git 的 .gitattributes 可以声明文件重命名、Rust 的 crate migration 规则、Kubernetes 的 API version 转换、TypeScript 的 @deprecated 标记它们都在解决"旧数据遇到新代码时怎么办"没有映射表、你的用户要么"永远升级不了"、要么"升级后数据全报废"——两种都不能接受映射表是"渐进式迁移"的技术基础LangChain 对这点的重视、体现了对生态演化的成熟思考

延伸阅读:ConfigurableField 的延迟绑定之美

ConfigurableField 让用户可以在运行时动态改变 Runnable 的配置——这是"延迟绑定"(late binding)设计的典范传统设计里、一个对象的行为在"创建时"就固定了——要改就要重新创建对象ConfigurableField 把这一部分延后到"调用时"——让同一个对象在不同调用下能有不同行为这对 A/B 测试、多租户、环境切换等场景极其有用

延迟绑定的思路、在其他系统也有——Python 的 duck typing(方法在调用时才解析)、JavaScript 的 prototype 链、依赖注入容器的 scope 管理——都是同一种思路它和"早绑定"(静态类型、编译时确定)形成对立——两者各有优劣早绑定:性能好、错误早发现;延迟绑定:灵活、适应动态需求好的系统会在两者之间按场景选择——LangChain 对运行时灵活性要求高、因此倾向延迟绑定;Rust 的类型系统对正确性要求高、因此倾向早绑定

延伸阅读:RunnableConfig 的传播机制

RunnableConfig 贯穿整个调用链——从顶层 Runnable 传到最底层的 tool call——中间不需要每一层显式传递这种"自动传播"的设计、让配置管理变得极其方便——你在顶层设置一次 callbacks 或 tags、所有子 Runnable 都会继承这是"context propagation"(上下文传播)模式的体现

上下文传播在其他系统也有——Python 的 contextvars、Node.js 的 AsyncLocalStorage、Rust 的 tokio::task_local、OpenTelemetry 的 trace context——都在解决"如何让跨函数调用保持某些值"的问题这种模式的关键、是"不用每个函数都显式传参"——减少了 boilerplate 代码、让业务逻辑更清晰LangChain 的 RunnableConfig 是这种模式的优雅应用——让 AI 应用的复杂调用链能保持一致的配置和观测性

延伸阅读:序列化与配置的哲学差异

本章最后把"序列化"和"配置"并列——它们看起来都是"让对象可调"、但哲学差异很大序列化是"对象状态"——把运行时的对象变成持久化的数据;配置是"运行时行为"——在调用时动态调整对象的行为两者正交——你可以序列化一个对象但不动它的配置、也可以配置一个对象但不序列化

这种"正交设计"是系统架构的高级境界让不同关注点互不干扰、让组合产生新可能——这比"一揽子解决"要优雅得多Unix 哲学里的"Do one thing and do it well"、微服务的"单一职责原则"、函数式编程的"关注点分离"——都是正交设计的不同表达LangChain 能在 AI 框架这个复杂领域里保持相对清晰、部分就来自这种"正交分解"的设计纪律——没有这种纪律、框架会迅速演化成不可维护的泥潭

延伸阅读:密钥序列化的安全陷阱

序列化的一个经典安全陷阱——"密钥也被序列化了"如果 ChatOpenAI 对象被 to_json 转成 JSON、然后保存到磁盘或传到其他地方——API Key 也跟着走了——这是一个极其危险的泄露路径LangChain 的解决方案——用 lc_secrets 声明哪些字段是敏感的、序列化时自动把这些字段的值替换为 SecretStr 占位符——真实密钥不出现在序列化结果里

这种"敏感数据自动脱敏"的机制、是面向生产应用的库必备特性类似的设计在其他库也有——Django 的 SECRET_KEY 不会被 settings 序列化、Spring Security 的 CredentialsContainer 可以 erase 敏感信息、Rust 的 secrecy crate 专门处理密钥类型——都是同一种"把敏感数据作为一等公民"的思路LangChain 的 lc_secrets 机制、是这种最佳实践在 AI 框架领域的具体应用

延伸阅读:序列化格式的选择

LangChain 选用 JSON 作为序列化格式——不是 pickle、不是 YAML、不是 Protocol Buffers为什么?因为 JSON 有几个关键优势——"跨语言"(Python 序列化的、Node.js 能读)、"人类可读"(debug 时能直接看)、"安全"(不像 pickle 那样能执行任意代码)、"广泛支持"(几乎所有语言和工具都原生支持)这些优势让 JSON 成为"跨进程、跨环境交换对象"的最佳选择

虽然 JSON 比二进制格式(Protocol Buffers、MessagePack)体积大、解析慢——但在 LangChain 的场景下、这些差异几乎可以忽略(序列化对象通常不大、也不是热点)"简单、通用、够用"往往比"最优性能"更重要——这是工程里的"80/20 原则"的又一次体现LangChain 的这个选择、体现了"务实优先"的工程价值观——先解决 90% 场景的需求、剩下 10% 靠扩展点

延伸阅读:配置系统对测试的影响

ConfigurableField 对测试极其友好——你可以给同一个 Agent 配置不同的 LLM、不同的 temperature、不同的 prompt——在测试时对比不同配置的表现这种"一套代码、多种配置"的测试模式、让 A/B 测试、性能对比、回归测试都变得简单没有 ConfigurableField、这些测试要么写一堆重复代码(每个配置一个版本)、要么改线上代码(风险高)

这种"为测试服务的设计"、是成熟库的标志很多库在设计时只考虑"功能好不好用"、不考虑"怎么测试"——结果用户测试痛苦、bug 频发LangChain 从第一天就考虑测试友好——ConfigurableField、RunnableConfig、Fake LLM、Callback 系统——都是为了让测试变简单这种"测试优先"的设计哲学、和 TDD(测试驱动开发)一脉相承——值得所有库作者学习

延伸阅读:序列化与对比诊断

序列化还有一个价值——"让不同版本的对象能对比"你可以把某个 Agent 在 v1 和 v2 的状态都序列化、然后 diff——看看具体哪里变了这对调试"为什么 v2 行为和 v1 不一样"极其有用——不用翻代码、直接看结构化 diff

这种"序列化支持诊断"的价值、在生产环境出问题时特别明显用户报"这个 Agent 以前好的、现在突然不对了"——你可以把两个版本的配置序列化出来对比——快速定位根因这和 Git 的 diff、数据库的 schema diff、Terraform 的 plan——都是同一类"让变化可视化"的工具LangChain 的序列化、除了"传输和持久化"这个显性价值、还有"诊断和对比"这个隐性价值——后者往往被低估、但在实际运维中价值巨大

延伸阅读:延迟绑定的成本

延迟绑定虽然灵活、也有成本——"性能开销"和"错误延迟"性能——每次调用都要查配置、比硬编码慢一点(通常可忽略)错误延迟——配置错误要等到运行时才暴露、不像静态类型那样编译时就报错LangChain 的 ConfigurableField 通过"运行时校验 + 友好错误消息"缓解了错误延迟问题——但无法完全消除

这种"灵活性和安全性的取舍"、是所有编程语言和框架都要面对的Python 选择灵活性(动态类型、延迟绑定)、代价是运行时错误;Rust 选择安全性(静态类型、早绑定)、代价是编译时间长没有完美答案——不同场景选不同方案LangChain 作为面向 AI 应用的框架、倾向于灵活性(因为 AI 场景变化快)——这是合理选择《Rust 编译器与运行时揭秘》第 8 章讨论的静态 vs 动态分发、《Vue 3 源码》第 5 章讨论的响应式 API 选择——都涉及类似的权衡、值得对照阅读

延伸阅读:持久化 Agent 的新兴价值

随着 Agent 应用越来越复杂——"持久化 Agent"(把完整的 Agent 配置 + 历史状态保存、以后继续用)成为一个有价值的能力LangChain 的序列化是实现这个能力的基础——没有序列化、Agent 一关就没了;有了序列化、Agent 可以"睡着"、再"醒来"

这种"Agent 持久化"、和"工作流持久化"(见《LangGraph 源码》第 8 章的 Checkpoint)类似——都在解决"让 AI 执行跨越时间"的问题未来可能出现的"Agent Marketplace"——用户可以分享自己训练调优好的 Agent 配置、其他用户下载直接用——这个愿景的技术基础、就是标准化的序列化格式LangChain 的序列化系统、为这个未来提前铺路——只是现在还没充分发挥价值——希望未来 2-3 年我们能看到"Agent 市场"这种新物种的诞生

延伸阅读:ConfigurableField 的命名技巧

ConfigurableField 里的 id 字段、是一个容易被忽略但极其重要的细节id 是配置字段的"外部名字"——用户在运行时改配置时、用这个名字指定改哪个字段好的 id 命名——"有明确含义"(temperature 而非 t)、"全局唯一"(避免和其他字段冲突)、"稳定不变"(不能因为重构就改)

这种"外部稳定接口 + 内部灵活实现"的设计、是库 API 设计的基本原则内部你可以随便改变量名、重构代码;但外部接口(包括 ConfigurableField 的 id)必须保持稳定、因为用户的代码依赖它违反这个原则的后果——每次版本升级都 break 用户代码——生态分裂、信任崩塌LangChain 在这方面做得相当好——id 稳定、用户升级时基本不用改配置代码——这种对"用户代码稳定性"的承诺、是长期生态健康的基础

延伸阅读:序列化的跨语言挑战

如果 LangChain 的 Python 版本序列化的对象、能被 LangChain.js 读——那就是真正的"跨语言互操作"这在理论上可能——因为两边都用 JSON 格式;实际上会遇到问题——类型映射(Python 的 datetime vs JS 的 Date)、命名空间(Python 的模块路径 vs JS 的包路径)、特有字段(某些 Python 对象的字段在 JS 实现里不存在)

LangChain 在跨语言互操作上做了一些工作——但还没达到"完全无缝"的水平这是一个"理论简单、工程困难"的典型问题——要真正跨语言、需要两边的团队协调、规范共同维护、测试矩阵覆盖——成本很高目前的状态是"95% 兼容、边缘 case 可能有问题"——对大多数用户够用、对精细互操作场景仍需调整这也提醒我们——"跨语言"听起来简单、实际做起来远非易事——需要持续的工程投入