LangChain可靠性：流式处理、批量处理、重试机制、备用方案与中间件

迄今为止，服务台客服人员所拥有的自由度其实相当有限。他们可以查看服务归属信息及近期发生的事件，但在撰写工单备注之前必须暂停，列出打算使用的依据，并等待人工介入。只要所有模型调用和工具请求都能及时响应，这种安排就能维持下去。但实际上并非如此。

一次模型请求可能需要15秒时间；服务提供商也可能因速率限制而将请求拒之门外长达一分钟；批量处理中的某张工单可能包含格式错误的输入数据；备用模型给出的回答可能与主模型不同；还有用户在已启动的工具运行过程中关闭浏览器。所有这些情况都应对应有明确的处理规则，并配有相应的测试。

这一步为客服人员增加了五项约束条件。针对单个请求而言：

one request
  -> stream honest progress
  -> retry a small transient failure
  -> use a labeled fallback if attempts are exhausted
  -> stop at time and call limits

针对一堆互不相关的请求：

many independent requests
  -> run a bounded batch
  -> keep one result or error beside each input

始终遵循的原则：不要用乐观的答复来掩盖故障。

我们持续用于测试的工单

测试场景保持不变：在14:05进行部署后，欧盟地区的结账请求会出现问题，结账API日志中会显示“数据库连接被拒绝”。用户要求客服查找工单负责人及近期相关故障，提出后续处理建议，并在用户确认后为INC-2048工单添加备注。

在理想情况下，客服会查看工单负责人及近期故障记录，撰写处理建议，提出备注内容，等待批准后一次性完成操作并反馈结果。为了让测试更复杂：同时提交三份工单，让某次供应商调用失败两次，使一份工单无效，并在首次请求过程中中断与客户的连接。以下所有设计都是为应对这类情况而制定的。

先给故障命名

“调用失败”这样的提示并没有告诉用户该怎么做。应将各种故障分类，以便采取不同的处理措施。

• 临时性服务提供商故障。可能是连接重置、请求超时，或是服务提供商明确要求稍后再试的回复。建议间隔一段时间后重试几次。
• 速率限制。服务提供商正在控制配额或处理能力。需按照其给出的重试说明操作，降低调用频率并减少并发数。立即重复尝试只会让情况更糟。
• 无效输入。工单缺少编号、结构不匹配，或文本长度超过了应用限制。应给出明确的客户端错误提示，切勿再次发送格式错误的请求。
• 授权失败。当前用户无权读取运行手册或向工单写入内容。此时应停止操作，既不要更换凭证，也不要让其他模型尝试绕过权限限制。

工具业务错误。工单已关闭，不允许进行状态转换，或存在相同的备注。应将工具的响应转换为有意义的应用程序错误，因为重试很可能会出错。

模型输出验证失败。服务提供者返回的数据违反了输出规范。对于只读结果，一次修复尝试可能是可接受的，但仍需有最终的失败处理路径。

未知错误。应用程序无法对错误进行分类。应标记为失败并记录错误类型以便进一步调查，切勿将所有异常都视为临时问题。

这些类别共同决定了哪些错误需要重试，哪些可以回退处理，以及哪些应直接反馈给调用方。

项目布局与设置

此步骤的快照包含：

langchain-helpdesk/
├── app.py
├── reliability.py
├── batch_runner.py
├── stream_view.py
├── middleware_config.py
└── tests/
    ├── test_reliability.py
    └── test_stream_view.py

安装这些软件包：

python -m pip install -U \
  langchain \
  langchain-core \
  langgraph \
  pytest

这些测试从不调用提供者。

流式处理负责传输数据，而非进行验证

在工具运行期间屏幕显示空白，对用户而言体验极差。流式处理通过在任务执行过程中发送事件来解决这一问题。LangChain当前的智能体流式功能提供了三种视图：

• updates：已完成的智能体步骤
• messages：模型生成的标记及元数据
• custom：由您的应用程序定义的进度数据

它们包含不同类型的信息，因此不应将所有内容视为同一答案来展示。

基本的流循环

for chunk in agent.stream(
    {
        "messages": [
            {
                "role": "user",
                "content": request_text,
            }
        ]
    },
    config,
    stream_mode=["updates", "custom"],
    version="v2",
):
    send_to_client(chunk)

详细解析：agent.stream(...)会启动与invoke(...)相同的执行流程，但在执行过程中会分批返回数据。第一个参数是状态更新内容，此处即一条用户消息。config用于存储与请求相关的设置，如线程ID，敏感信息不应放入其中。stream_mode=[“updates”, “custom”]会要求返回已完成步骤的数据以及自定义的进度消息，因此在本例中不会按字符逐个返回文本。version=“v2”指定了当前的块格式，每个块包含类型、命名空间和数据载荷。send_to_client(chunk)指的是用户自定义的代码，它可以发送服务器推送事件、WebSocket消息或终端行。

为界面翻译原始数据块

该功能会在数据块到达用户界面之前对其进行转换：

@dataclass(frozen=True)
class ProgressEvent:
    kind: str
    label: str
    payload: Any

def readable_events(chunks):
    for chunk in chunks:
        chunk_type = chunk.get("type")
        data = chunk.get("data")        if chunk_type == "updates" and isinstance(data, dict):
            for step_name, step_data in data.items():
                yield ProgressEvent(
                    kind="step",
                    label=step_name,
                    payload=step_data,
                )

该适配器为用户界面提供了简洁且稳定的接口规范。它能够显示如下内容：

Checking service ownership...
Reading recent incidents...
Waiting for ticket-note approval...
Ticket note approved and written.

在运行实际结束之前，绝不能显示以下内容：

Done.

部分内容并非最终结果

流式处理机制虽能让响应显得更快，但并不会提升安全性或准确性。模型可能会在写到一半就跳转到其他内容，工具也可能在令牌已发送后出现故障，此外策略检查还可能阻止拟执行的操作。

应使用明确的事件状态：running、waiting_for_approval、completed、failed和cancelled。只有completed状态才代表最终结果。

明确断开连接的含义

浏览器关闭并不代表服务器已停止运行。需提前制定相关策略：

• 若存在取消操作且无人正在等待结果，则终止只读任务。
• 如果用户计划稍后继续处理，就让已获批准的持久任务完成。
• 绝不能仅因为写入操作的响应丢失就认为该操作并未执行。

在再次尝试写入之前，先通过幂等键查找该操作。

流式处理与取消操作相关，但并非同一功能。

独立输入的批量处理

LangChain 可执行组件提供了 batch 和 abatch 函数。默认的 batch 会在线程池中执行独立的调用，这适用于许多受网络限制的操作；而特定提供者的可执行组件则可用原生批量 API 取代这一方式。服务台工作人员会接收多份互不相关的工单：

outputs = runnable.batch(
    tickets,
    config={"max_concurrency": 4},
    return_exceptions=True,
)

tickets 是输入项的有序列表。max_concurrency 用于限制同时运行的调用数量，4只是一个示例值；实际数值应根据服务提供商的配额、工具的处理能力、成本以及实测延迟来确定。当设置 return_exceptions=True 时，即使某个输入项处理失败，也不会影响其他输入项的结果。之后，应用程序会将每个输出结果与其对应的输入项匹配起来：

for ticket, output in zip(tickets, outputs, strict=True):
    if isinstance(output, BaseException):
        save_failure(ticket.ticket_id, type(output).__name__)
    else:
        save_recommendation(ticket.ticket_id, output)

这样每个工单只会产生一个结果。

在以下情况下应跳过并行批处理：第二项依赖于第一项；所有项目都在没有并发控制的情况下更新同一条记录；执行顺序本身即为一条业务规则；服务提供方采用具有不同保障级别的批量事务处理；或者某个用户提交的项数足够多，以至于占用了所有共享资源。批处理只是分散负载，并不能使其消失。

在最小的安全范围内重试

重试意味着要再次执行相关操作，因此需谨慎选择重复的内容。重新运行整个工作流会再次进行数据检索、模型调用以及工单写入操作。更安全的方法是仅对已知会暂时出错的读取步骤进行重试。为此，快照机制专门定义了相应的错误处理方式：

class TemporaryProviderError(RuntimeError):
    """A short-lived error that may succeed when tried again."""

并包裹一个可运行的代码块：

primary_call = RunnableLambda(primary).with_retry(
    retry_if_exception_type=(TemporaryProviderError,),
    wait_exponential_jitter=False,
    stop_after_attempt=3,
)

RunnableLambda(primary)可将普通的Python函数转换为可执行的对象。retry_if_exception_type起允许列表的作用，无效的请求不会被匹配，会直接返回给调用方。设置wait_exponential_jitter=False可使离线测试更快且结果更确定。stop_after_attempt=3表示总共进行三次尝试，而非在第一次尝试基础上再重试三次。针对真实网络服务，采用带抖动的指数退避策略通常是更好的选择，因为这样能避免所有工作线程同时访问该服务。

写入操作不可盲目重试

想象一下，票务服务添加了该备注，但在成功响应到达之前连接就中断了。调用方会看到超时错误，但实际上该写入操作可能已经完成，若简单重试就会导致备注被重复添加。此时应使用幂等性键：

operation_id = "ticket-note:INC-2048:approval-7331"

票务服务会在首次成功写入时保存该键值。如果相同的键再次出现，它会返回已存储的结果而不会重复应用更改。当外部系统无法实现幂等性时，应在重试前检查其当前状态，并接受某些故障窗口无法完全消除的事实。

备用方案是另一条路径

一旦允许的尝试次数用尽，备用可执行程序便可接管任务：

reliable = primary_call.with_fallbacks(
    [RunnableLambda(conservative_fallback)],
    exceptions_to_handle=(TemporaryProviderError,),
)

它仅针对指定的临时错误运行，其返回结果如下：

Recommendation(
    ticket_id=ticket.ticket_id,
    text=(
        "The recommendation service is temporarily unavailable. "
        "Keep the ticket open and compare the failed deployment "
        "with the last known-good release."
    ),
    source="fallback",
)

该回复承认了所发生的情况，但并未声称曾搜索过操作手册或更新过工单。

切换到另一个模型并不等同于重启第一个模型。第二个模型在调用工具时的表现可能不同，上下文长度限制更小，对结构化输出的支持方式也有所差异，还会采用其他安全机制，同时在延迟和成本方面也存在区别，还可能受制于不同的供应商数据政策。应将备用路径视为独立的配置，并单独进行测试。最好的备用方案有时根本不用模型，直接使用固定的保守性消息或将其放入队列等待人工审核。

超时机制存在于多个层面

一个超时值是不够的。应为连接建立、单个模型请求、单个工具请求、整个代理运行过程、审批等待以及客户端连接分别设置预算。整体运行预算必须高于任何单个请求的预算，但仍需设定上限。

应将超时机制设置在它所限制的操作旁边：在提供方客户端设置提供方超时，在工具客户端设置工具超时，而在服务边界则设置网页请求的截止时间。一旦触发超时，需记录是哪一层终止了该操作。仅出现“超时”字样而没有明确说明触发层级的状况很难解决。

中间件封装代理调用

前面的步骤主要使用固定链，而审批步骤则引入了代理中间件。对于那些需要覆盖多种模型或工具调用的规则，中间件也是一个理想的载体：例如重试特定的模型失败请求、限制模型调用次数、限制工具调用次数、为当前用户筛选可用工具、为日志和追踪添加请求元数据、隐藏特定的个人数据，以及在应用预算耗尽时停止运行。该快照使用了内置的中间件：

def reliability_middleware():
    return [
        ModelRetryMiddleware(
            max_retries=2,
            retry_on=(
                TemporaryProviderError,
                TimeoutError,
                ConnectionError,
            ),
            on_failure="error",
            initial_delay=0.5,
            backoff_factor=2.0,
            max_delay=4.0,
            jitter=True,
        ),
        ModelCallLimitMiddleware(
            run_limit=8,
            exit_behavior="error",
        ),
        ToolCallLimitMiddleware(
            run_limit=6,
            exit_behavior="end",
        ),
    ]

ModelRetryMiddleware用于包裹模型调用，初始调用加上两次重试最多进行三次尝试。ModelCallLimitMiddleware可防止模型循环无限运行。ToolCallLimitMiddleware则能限制单次运行中工具的使用频率。这些机制各司其职：调用次数限制不能替代超时处理，超时处理不能替代审批流程，而审批流程也不能替代授权机制。

顺序很重要

在调用之前执行的钩子按列表顺序排列，之后执行的则按相反顺序排列，且嵌套的钩子会像函数调用那样层层嵌套。当中间件A包裹了中间件B时，A中的重试操作可能会重复执行B的任务，因此每当有两个组件都可能进行重试、修改模型或改变状态时，都需要明确其执行顺序。写入操作的审批流程应置于任何可能重复该已批准操作的重试机制之外。

自定义钩子就是普通代码

LangChain提供了节点式钩子（before_model、after_model）和包装式钩子（wrap_model_call、wrap_tool_call）。节点式钩子适用于在固定节点处进行检查与状态更新；而包装式钩子则用于需要包裹某次调用的代码，或是决定处理程序应执行零次、一次还是多次的逻辑。不过，并非所有的应用规则都适合转化为中间件——任务状态转换、授权决策以及数据所有权规则作为命名后的应用服务来处理会更为清晰。

离线测试故障路径

该快照包含五个测试，均无需提供者。

临时错误在第二次尝试时能够成功处理：

result = runnable.invoke(ticket())

assert calls == 2
assert result.source == "primary"

当尝试失败时，将使用标记好的备用方案：

assert calls == 3
assert result.source == "fallback"
assert "temporarily unavailable" in result.text

无效输入既不会被重试也不会被隐藏：

with pytest.raises(InvalidTicketError):
    runnable.invoke(ticket())

assert calls == 1

一次批次失败并不会抹去另一次的成功：

assert results[0].status == "completed"
assert results[1].status == "failed"
assert results[1].error_type == "RuntimeError"

进度与模型令牌是分开的：

assert [(event.kind, event.label) for event in events] == [
    ("step", "model"),
    ("progress", "custom"),
    ("token", "model"),
]

运行它们：

pytest -q

预期结果：

5 passed

这些测试能够验证你的本地控制规则，但无法证明服务提供商的错误已被正确映射，这需要通过真实的提供商SDK进行简单的集成测试，并使用受控的测试账户来完成。

记录故障矩阵

在发布之前，应在代码或文档中放入如下表格：

failure                         retry?   fallback?   client result
connection reset               yes      maybe       503 if exhausted
provider rate limit            yes      maybe       503 or queued
invalid ticket                 no       no          422
authorization denied           no       no          403
structured output invalid      limited  maybe       502 if exhausted
ticket already closed          no       no          409
unknown exception              no       no          500 + incident ID

具体的状态码取决于您的 API。重要的是，每一种故障类型都有明确的处理策略。

仍存在不足

目前的运行时行为已经更加清晰，但这仍然只是一个命令行层面的演示版本。它没有经过身份验证的 HTTP API，没有启动和就绪状态检查，没有密钥加载机制，没有结构化的请求日志，没有持久化的任务存储功能，没有部署配置选项，也没有真正的供应商集成测试，发布前没有评估集校验，更没有生产环境下的追踪审查。具备这些功能的服务版本将在后续推出。

客服处理流程的现状

目前的流程具有可见的状态以及有限的恢复能力：

request
  -> progress events
  -> small retry boundary
  -> labeled fallback when allowed
  -> model and tool call limits
  -> approval before write
  -> final result or named failure

batch
  -> bounded concurrency
  -> one outcome per ticket

没有任何控制手段能够完全消除不确定性。其目标是将这种不确定性保持在可见、可控且可检测的范围内。接下来是API背后的完整服务台工作流程，包括配置、密钥、请求ID、超时设置、日志记录以及追踪机制；随后是确定性测试、评估用例和部署检查，还有在服务处理实际工单之前企业必须提供的各项内容清单。以上所有内容均已于2026年8月31日根据LangChain关于流式处理、可重试机制、备用方案、批量处理、中间件以及测试的Python文档进行了核对，而相关API包及服务提供商的错误类型可能会发生变化。