自定义链

🔗 1. 如果要自定义一条链，需要继承什么类？必须实现哪些方法？

最正统的做法是继承 Chain 基类（from langchain.chains.base import Chain）。早期版本里大家也会直接继承 LLMChain 或者 SimpleSequentialChain 去魔改，但现在有 Runnable 协议之后，我更建议在思想上先理解：链的本质就是一个可运行对象，有输入、有输出、有内部逻辑，并且能被组合。

但直接回答你的问题：继承 Chain 时，必须实现的硬性方法其实只有两个：

• _call：同步执行的核心逻辑，接收 dict 输入，返回 dict 输出。

• @property 的 input_keys 和 output_keys：告诉框架你这条链吃哪些键、吐哪些键。

为什么说“只有两个”？因为很多人会提到必须实现 _acall，但从技术上讲，如果你不实现它，调用 ainvoke 时会自动用 run_in_executor 把同步的 _call 包成异步，虽然不推荐，但不会报错。不过，真要上生产，你必须把 _acall 老老实实写好，这个后面再细说。

还有一个经常被忽略但非常重要的点：如果你想用 chain.invoke 这种统一接口，你需要知道在 LangChain 0.1.x 之后，自定义 Chain 最好同时混入 Runnable 的特性。更现代的做法是直接实现 Runnable 协议，而不是死守旧 Chain 类。不过面试场景中，面试官通常还是问旧 Chain 那一套，所以我会这样回答：“可以继承 Chain，必须实现 _call 方法和 input_keys/output_keys 属性。如果你还需要更灵活的流式、事件回调等控制，建议了解 RunnableLambda 或者直接实现 Runnable 协议，那里你只需要实现 invoke 和可选的 ainvoke 等。”

我个人习惯在自定义链时还会重写 _chain_type 属性返回一个自定义字符串，这在监控和日志里非常有用，一眼就知道是哪个链出了问题。

---

🧪 2. 写一个自定义链，输入是一段文本，输出是这段文本的英文翻译和情感分析结果。

我先给个具体实现，然后再说设计上的取舍。假设我们底层有两个能力：翻译用某个 LLM，情感分析用另一个轻量模型或同一个 LLM 的不同 prompt。代码可以这样写：

from typing import Dict, Any, List
from langchain.chains.base import Chain
from langchain_core.language_models import BaseLLM
from langchain_core.prompts import PromptTemplate

class TranslateAndSentimentChain(Chain):
    """输入一段文本，输出英文翻译和情感分析结果"""

    llm: BaseLLM
    translate_prompt: PromptTemplate = PromptTemplate(
        input_variables=["text"],
        template="将以下文本翻译成英文，只返回英文译文：\n{text}"
    )
    sentiment_prompt: PromptTemplate = PromptTemplate(
        input_variables=["english_text"],
        template="分析下面英文文本的情感，只输出一个单词（positive/negative/neutral）：\n{english_text}"
    )

    @property
    def input_keys(self) -> List[str]:
        return ["text"]

    @property
    def output_keys(self) -> List[str]:
        return ["english_translation", "sentiment"]

    def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
        text = inputs["text"]

        # 第1步：翻译
        translate_input = self.translate_prompt.format(text=text)
        translation = self.llm.invoke(translate_input).strip()

        # 第2步：情感分析
        sentiment_input = self.sentiment_prompt.format(english_text=translation)
        sentiment = self.llm.invoke(sentiment_input).strip().lower()

        return {
            "english_translation": translation,
            "sentiment": sentiment
        }

这个实现很直白，但面试官这时候很可能追问：“你这里用的是同一个 LLM 实例，如果翻译和情感分析需要不同模型呢？” 那就要把 translate_llm 和 sentiment_llm 分开传入。实际项目中我确实这么干过，翻译用 GPT-4 保证质量，情感分析用 GPT-3.5 省钱，当时就在构造函数里区分了 translate_llm 和 sentiment_llm，并且每个 prompt 都单独暴露为可配置参数，这样当产品经理一会儿要翻译成德语、一会儿要输出“愤怒/惊喜”这种多分类情感时，我只需要改配置，不用动链的逻辑。

另外，如果让我现在重新设计，我会让这个链直接实现 Runnable，用 RunnableParallel 和 RunnableLambda 组合，因为这样可以把翻译和情感分析的依赖关系显式化，还能自动获得并行能力（虽然这里有依赖，不能完全并行）。但为了展示对传统 Chain 的掌握，上面这种写法足够了。

---

🛡️ 3. 在自定义链中，如何保证输入输出的 Schema 验证？如何定义 input_schema 和 output_schema？

Schema 验证是自定义链能不能在复杂工作流里活下来的关键。没有 Schema 验证，一条链默默吞了错误 key，后面链路就会崩得莫名其妙。LangChain 提供的方式是重写 input_schema 和 output_schema 这两个类方法，它们返回 Pydantic 的 BaseModel。

from pydantic import BaseModel, Field

class TranslateAndSentimentChain(Chain):
    # ... 前面属性 ...

    @classmethod
    def input_schema(cls) -> type[BaseModel]:
        class InputSchema(BaseModel):
            text: str = Field(..., description="需要分析和翻译的原始文本")
        return InputSchema

    @classmethod
    def output_schema(cls) -> type[BaseModel]:
        class OutputSchema(BaseModel):
            english_translation: str = Field(..., description="英文翻译结果")
            sentiment: str = Field(..., description="情感分析结果（positive/negative/neutral）")
        return OutputSchema

但注意，旧版 Chain 的 validate_inputs 和 validate_outputs 是内部方法，你只要定义了 schema，框架在 invoke 时会自动基于 Pydantic 做校验。我在早期踩过一个坑：输入 key 叫 input_text，但我 schema 里定义的是 text，结果调用时没报错，链内部拿了 inputs["text"] 直接 KeyError，排查了好久。后来学乖了，所有自定义链都严格定义 schema，并且配合单元测试校验。

实际经验里，如果你的链被用于 SequentialChain 或者 RouterChain，output_schema 的字段会作为后续链 input_mapping 的依据，所以一定要把输出字段的类型和描述写清楚。有时我甚至会在 output_schema 里用 validator 做进一步的格式检查，比如确认情感分析结果真的只在那三个词里，不过这会增加运行时开销，看场景权衡。

还有一个细微点：input_schema 和 output_schema 更多是给外部系统（比如 LangServe、Tracing、文档生成）看的，框架实际运行时不强制用它们校验，除非你显式调用 validate_inputs。我一般在 _call 开头自己加一句 self.validate_inputs(inputs) 来提前暴露问题，这是一个防御性编程的习惯。

---

🔄 4. 当自定义链内部调用了其他链或模型，如何把 callback 和配置正确地传递下去？

这是一个老生常谈但非常容易出错的点。很多初学者在自定义链里直接 self.llm("some prompt")，结果发现 LangSmith 追踪里断掉了，或者回调函数根本没触发。原因就是 callback 和配置（如 tags、metadata）不会自动向下渗透，你需要显式传递。

在 _call 方法中，框架会把一个 CallbackManager 对象放到 inputs 旁边的运行上下文中，你可以通过 self.callback_manager 拿到，它内部有 on_llm_start 等方法。当你在链里调用子链或模型时，需要手动把 callback 传过去：

def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
    # 获取回调管理器
    cb = self.callback_manager
    text = inputs["text"]

    # 翻译时传入回调
    translate_input = self.translate_prompt.format(text=text)
    translation = self.llm.invoke(
        translate_input,
        config={"callbacks": cb}  # 关键点
    ).strip()

    # 情感分析同理
    sentiment_input = self.sentiment_prompt.format(english_text=translation)
    sentiment = self.llm.invoke(
        sentiment_input,
        config={"callbacks": cb}
    ).strip().lower()

    return {...}

但是！如果你的自定义链还接收了 run_name、tags、metadata 这些运行时配置，你也应该一并往下传，不然你在大链路里加了全局 tag “production”，结果内嵌的 LLM 调用却没有带上这个 tag，出问题时你就没法按标签过滤日志了。我通常的做法是直接从 self 提取所有运行时配置，构造成 RunnableConfig 字典传入：

config = {
    "callbacks": self.callback_manager,
    "tags": self.tags,
    "metadata": self.metadata,
    "run_name": self.run_name,  # 有时给子调用单独命名
}

另一个容易被忽略的是 异步调用 时的传递：_acall 里一样要手动传，否则异步链路追踪也会断。我习惯封装一个 _get_runtime_config() 的私有方法，在同步和异步里复用，保持一致性。

另外，如果你的子组件是 LangChain Expression Language (LCEL) 构建的 Runnable 对象，它的 invoke 方法本身就接受 config，这时候直接把上面拼好的 config 字典传进去就行，回调就会在整个 Runnable 拓扑中自动传播——这才是 LCEL 的优雅之处。所以现在新项目里，我更倾向于把内部逻辑也用 LCEL 写，传一次 config 就一劳永逸。

---

⚡ 5. 你的自定义链需要支持异步调用，该如何实现 _acall 方法？

首先声明：如果不写 _acall，LangChain 会用默认实现把你的同步 _call 丢进线程池模拟异步，这在 IO 密集型且并发不高时勉强可用，但一旦你要在异步 Web 框架（如 FastAPI）里用，会阻塞事件循环，性能完全不行。所以生产环境必须自己实现 _acall。

实现的关键在于：你要保证所有内部调用都走异步路径。如果你用的是 LLMChain 或直接调 llm.apredict()，那没问题；但如果你内部调了一个旧版的只支持同步的链，那你需要在 _acall 里用 asyncio.to_thread 把那部分包出去，否则会阻塞。

一个典型的 _acall 实现长这样：

async def _acall(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
    text = inputs["text"]
    # 获取运行时配置
    config = {
        "callbacks": self.callback_manager,
        "tags": self.tags,
        "metadata": self.metadata,
    }

    # 异步翻译
    translate_input = self.translate_prompt.format(text=text)
    translation = (await self.llm.ainvoke(translate_input, config=config)).strip()

    # 异步情感分析
    sentiment_input = self.sentiment_prompt.format(english_text=translation)
    sentiment = (await self.llm.ainvoke(sentiment_input, config=config)).strip().lower()

    return {
        "english_translation": translation,
        "sentiment": sentiment
    }

如果内部两个调用没有依赖关系，你可以用 asyncio.gather 并行加速，但这里情感分析依赖翻译结果，只能串行。我之前做过一个链需要同时调用三个外部 API 做交叉验证，在 _acall 里用 asyncio.gather 并发请求，响应时间直接降到三分之一，非常爽。

还有一个小技巧：如果你的 llm 本身就是 AsyncLLM 的子类，ainvoke 会走真正的异步 IO；但如果是同步 LLM 被包装了，那实际上还是在跑线程。所以如果你的链运行在高并发异步环境，请确保底层模型对象支持原生异步，比如 ChatOpenAI 的 async_client。

另外，写 _acall 时千万别忘了把 callback_manager 也切换成异步模式（其实 self.callback_manager 在异步上下文里已经是 AsyncCallbackManager 了，只要你在 _acall 里用 await 而不是同步方法，回调就能正确记录异步事件）。简单来说，照着同步逻辑写，把所有 .invoke 换成 await .ainvoke，显式传递 config，十有八九不会翻车。

---

🛟 6. 在自定义链中，如何处理可能发生的异常，并返回一个安全的默认结果？

异常处理是自定义链从“能跑”到“可靠”的分水岭。我的原则是：不要让异常裸奔出链边界。在 _call 或 _acall 内部，应该捕获所有可预见的异常，并按照业务需求返回一个预定义的 fallback 结果，同时记录足够多的上下文日志。

最常见的两种异常：

1. LLM 调用失败：网络超时、限流、内容审核触发等。

2. 输出解析失败：模型返回格式不符合预期，比如情感分析没返回那三个词之一。

对于翻译加情感分析的链，我会这样处理：

def _call(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
    text = inputs["text"]
    try:
        translation = self.llm.invoke(...).strip()
    except Exception as e:
        logger.error(f"翻译失败: {e}", extra={"input_text": text})
        translation = "[Translation Failed]"

    try:
        sentiment = self.llm.invoke(...).strip().lower()
        if sentiment not in ("positive", "negative", "neutral"):
            sentiment = "neutral"  # 输出修正
    except Exception as e:
        logger.error(f"情感分析失败: {e}", extra={"translation": translation})
        sentiment = "neutral"

    return {
        "english_translation": translation,
        "sentiment": sentiment
    }

注意，这里我用了“单个步骤独立 try-except”的策略，因为翻译失败不意味着整个链要崩，我们还可以用原文做点什么，或者至少返回一个失败占位符，让上游调用方决定怎么处理。如果整体失败，我会在返回的 dict 中增加一个 error 字段，形成一个约定：{"english_translation": "...", "sentiment": "...", "error": None}，当 error 不为 None 时，上游就知道部分数据不可靠。

还有一点，异常不要悄悄吞掉。你必须在日志里留下完整的堆栈，并且最好把失败的输入也记录下来，方便后续复现和修复。如果你的链用在自动化流水线里，甚至可以接入告警，当某类异常出现频次超过阈值时主动通知。

对于 _acall，记得用 try-except 包裹 await 调用，原理相同，只不过异步里的异常堆栈有时会丢失，可以用 logger.exception 来记录完整上下文。

安全默认值的选择取决于业务。做情感分析，默认 neutral 影响最小；做翻译，返回失败标记而不是空字符串，能避免下游把空串当正常英文去处理，进而引发更诡异的 bug。

---

💾 7. 如何在自定义链中加入缓存机制，避免相同输入重复调用 LLM？

缓存是降本增效的大杀器，尤其在调试或重复请求的场景下。LangChain 本身提供了 InMemoryCache、SQLiteCache、RedisCache 等，通过 llm.cache 属性设置即可。如果你的自定义链只用了一个 LLM，那直接开启 llm.cache = True 或者传入一个 Cache 对象是最简单的：

from langchain.cache import InMemoryCache
self.llm.cache = InMemoryCache()

但更常见的情况是：你的链内部可能用了多个模型，或者你希望基于整个链的输入输出做缓存，而不是每个 LLM 调用的缓存。这时候你可以自己实现缓存逻辑。我习惯在 _call 方法里用 hash 或 json.dumps 生成输入指纹，然后查缓存：

import hashlib, json
from langchain_core.caches import BaseCache

class TranslateAndSentimentChain(Chain):
    cache: BaseCache = None   # 让外部注入

    def _call(self, inputs):
        if self.cache:
            key = hashlib.md5(json.dumps(inputs, sort_keys=True).encode()).hexdigest()
            cached = self.cache.lookup(key, "TranslateAndSentimentChain")
            if cached:
                return json.loads(cached)  # 注意反序列化
        # ... 正常逻辑 ...
        result = {...}
        if self.cache:
            self.cache.update(key, "TranslateAndSentimentChain", json.dumps(result))
        return result

但要注意，缓存键的设计需要考虑可变因素，比如 prompt 模板、模型参数是否会影响结果。如果你可能调整 prompt，那缓存键就要把 prompt 的 hash 也算进去，否则改了 prompt 但文本一样，命中了旧缓存，产出错误结果。

还有一种更工程化的做法：使用 LangChain 的 RunnableWithCache 包装你的链，这需要你的链是 Runnable。那么你可以用 RunnableLambda 包一下，然后套 RunnableWithCache，它内部自动处理键的生成和缓存查询，并支持 TTL、后端选择等。我在一个内部工具里就用 RedisCache 做了全局缓存，然后所有 LLM 调用和链调用都经过这个缓存，缓存命中率很高，成本下降明显。

不过，缓存也有副作用：情感分析这种偏主观的任务，当模型微调或版本升级后，旧缓存可能不再合适，所以一定要设计缓存失效策略，要么在键里带上模型版本，要么提供手动刷新接口。

---

🔌 8. 自定义链中如果需要访问外部资源（如数据库、API），你一般怎么初始化？在链的构造函数中传入吗？

这是一个设计哲学问题。我的原则是：依赖注入 > 内部硬编码。所以，访问外部资源（数据库连接、API 客户端、Redis 客户端等）都应该通过构造函数传入，而不是在链内部自行创建。

理由很充分：

1. 测试友好：测试时你可以注入一个 mock 的数据库连接或假 API 客户端，而不用真的连外部服务。

2. 配置集中管理：连接字符串、超时时间、池大小等配置集中在应用启动时处理，链本身无需关心。

3. 资源生命周期：由外部管理连接池的开启和关闭，链不负责这些，避免资源泄漏。

4. 复用：多个链实例可以共享同一个数据库连接池。

所以我会把外部资源定义为 Pydantic Field，类型用 Any 或具体的协议类：

class MyChain(Chain):
    db_session: Any          # 可以是 sqlalchemy session
    api_client: SomeApiClient
    cache_client: Any = None
    # ...

然后在 _call 里直接使用 self.db_session.query(...)。这里要注意线程安全：如果你的链被用在多线程环境，而数据库 session 不是线程安全的，那就要在每次 _call 里创建新 session，但可以通过注入一个 session_factory 来避免硬编码：

class MyChain(Chain):
    session_factory: Callable[[], Session]

    def _call(self, inputs):
        with self.session_factory() as session:
            # 使用 session

对于异步资源（比如 httpx.AsyncClient），一样通过构造函数注入，但需要确保调用方在应用生命周期内管理这个 client 的开启和关闭。如果链需要自己初始化一个轻量资源（比如每次调用都新建一个临时 API 客户端），那也可以，但要明确这个资源不是共享的，并且要注意开销。

一个反例：我见过有人在自定义链的 init 里直接 self.db = create_engine(...)，导致测试时疯狂连生产数据库，最后不得不 mock 整个类，非常痛苦。所以，务必依赖注入。

---

🧪 9. 如何测试自定义链？你有什么策略来 mock 内部的 LLM 调用？

自定义链的测试可以分为三层：单元测试、集成测试和端到端测试。对于有 LLM 调用的链，单元测试的核心就是 mock 掉 LLM，保证你的逻辑逻辑（prompt 拼接、条件分支、输出组装等）正确。

我最常用的策略是使用 unittest.mock 或者 pytest-mock 来 mock 底层 LLM 的 invoke 和 ainvoke 方法。比如：

def test_translate_and_sentiment():
    mock_llm = MagicMock()
    mock_llm.invoke.side_effect = [
        "Hello, how are you?",   # 第一次调用：翻译结果
        "positive"               # 第二次调用：情感分析结果
    ]
    chain = TranslateAndSentimentChain(llm=mock_llm)
    result = chain.invoke({"text": "你好，你怎么样？"})
    assert result["english_translation"] == "Hello, how are you?"
    assert result["sentiment"] == "positive"
    assert mock_llm.invoke.call_count == 2

如果链内部用了多个 LLM 实例，就分别 mock。side_effect 的顺序要和调用顺序严格一致，这个在编写测试时要注意。

对于更复杂的链，我会使用 FakeLLM（LangChain 提供的测试用假模型）或者自己实现一个 FakeChatModel，它根据输入 prompt 中的某个标志返回预定输出，这样链的复杂分支也能测到。

测试异步链时，记得把 mock_llm.ainvoke 设成 AsyncMock，然后用 pytest.mark.asyncio 来跑。

集成测试方面，我会搭一个本地的小型服务，比如用 vcrpy 录制真实的 LLM API 响应，然后回放，这样能测到 prompt 是否真的符合模型预期格式，而又不浪费调用额度。还有一种做法是部署一个本地 LLM（如 Ollama 或 LocalAI）做集成测试，缺点是慢一点，但能发现很多网络层面或解析层面的问题。

测试 Schema 验证：写一个测试用例，故意传入不符合 schema 的输入，预期抛出 ValidationError；输出也可以用 chain.output_schema 来校验返回的 dict。

测试异常处理：mock LLM 让 invoke 抛出 TimeoutError，检查是否返回了 fallback 结果并记录了日志。

我的经验是，链的测试重点在于逻辑编排，而不是模型效果，所以 mock 越轻量越好。同时，每一个自定义链都必须有一条“快乐路径”测试和至少两条“失败路径”测试，这样重构时才不会胆战心惊。

---

💡 10. 你如何看待 LangChain 中“一切皆链”的思想？它会不会导致滥用？

我先亮观点：“一切皆链”在早期确实降低了编排大模型工作流的门槛，但现在看来，它也有被滥用的风险，需要辩证看待。

积极的方面： 链的抽象统一了接口，把 LLM 调用、工具使用、数据处理都变成可组合的单元。比如 prompt | llm | output_parser 这种管道，读起来就像流水线，非常符合直觉。对于快速原型和简单流程，链确实极大提高了开发效率。而且，这种“一切皆链”的思想让追踪、回调、批处理、流式输出等能力可以无差别地应用到每个组件上，这是非常优雅的设计。

滥用的风险：

1. 过度封装：很多人把一段简单的 if-else 逻辑也写成一个自定义 Chain，本来几行 Python 函数就能搞定的事，硬是套了一层 Chain 的壳，引入了大量隐式行为和 Pydantic 校验开销，代码反而难读、难调试。

2. 性能陷阱：链在内部有很多字典拆包、键映射、回调传播等隐式操作，如果不了解这些开销，把大量细粒度操作都封成 Chain，可能导致不必要的序列化、反序列化和字典拷贝，在高吞吐场景下问题会暴露。

3. 调试心智负担：一个过于复杂的链（尤其是嵌套了很多层的 SequentialChain 或 RouterChain），出错时你需要在脑海中还原整套链的拓扑，单步调试也异常困难，因为调用栈里全是框架代码。

4. 过早抽象：在业务逻辑没有稳定之前，用链强行统一所有流程，后期一旦需求变化（比如从顺序变成并行，或者条件分支变多），修改成本很高，还不如用普通函数组合。

我现在的实践是：

• 简单的、独立的功能直接用 RunnableLambda 包装一个普通 Python 函数，不刻意去继承 Chain。

• 多个步骤需要组合，并且需要追踪、并发控制、错误边界时，用 LCEL 的 RunnableParallel、RunnableBranch 等构建 DAG，而不是用旧式 Chain 的子类。

• 自定义链（继承 Chain）只用在需要复杂内部状态、生命周期管理（如资源初始化）或者需要暴露特殊配置的场景。

• 无论用什么形式，都保持组件职责单一，链的嵌套深度不超过 2 层。

总之，不要把“一切皆链”当成信仰。它是一个工具，当它让你的代码更清晰、更容易测试和维护时，它就是好的；当它开始掩盖逻辑、制造混乱时，就应该果断用最朴素的 Python 类或函数替代。这个思想本身没错，关键是使用者的工程判断力。

---