---
description: "使用 dowhen 库在运行时对第三方库的方法进行精确补丁，无需 fork 或等待上游修复"
---

# 运行时补丁 (dowhen)

使用 dowhen[^dowhen] 库在运行时对第三方库的方法进行精确补丁，无需 fork 或等待上游修复。

## 适用场景

- 第三方库的某个方法有 bug，但不想 fork 整个库
- 需要在特定条件下改变第三方库的行为
- 临时兼容性修复（等上游合并 PR 期间的过渡方案）

## 基本语法

```python
from dowhen import when

# 在 SomeClass.some_method 中，找到 'target_line' 这行代码
# 当 condition 为真时，用 .do() 中的代码替换/插入
(
    when(
        SomeClass.some_method,
        'target_line_of_code',        # 定位到方法中的哪一行
        condition='some_condition'     # 可选：仅在条件成立时执行
    )
    .do("replacement_code")
)
```

## 实际案例

### 案例1：修改 Langfuse 回调的日志级别

当 LangGraph 流被主动中断（GeneratorExit）或前置检查异常时，Langfuse 会记录 ERROR 级别。用 dowhen 改为不记录：

```python
from langfuse.langchain.CallbackHandler import LangchainCallbackHandler
from dowhen import when

(
    when(
        LangchainCallbackHandler.on_chain_error,
        'observation = self._detach_observation(run_id)',
        condition=(
            'isinstance(error, GeneratorExit)'
            'or "PreconsicousError" in str(type(error))'
            'or "CancelledError" in str(type(error))'
        )
    )
    .do("level = None")  # 将 level 设为 None，跳过错误记录
)
```

### 案例2：修改工具错误的记录方式

将特定业务异常的工具错误从 ERROR 降级为 WARNING：

```python
do2 = '''
if observation is not None:
    observation.update(
        status_message=str(error),
        level="WARNING",
        input=kwargs.get("inputs"),
        cost_details={"total": 0},
    ).end()
observation = None
'''

(
    when(
        LangchainCallbackHandler.on_tool_error,
        'if observation is not None:',
        condition='(error.__class__.__name__ in ["ChangeToolError"])'
    )
    .do(do2)
)
```

### 案例3：兼容 DeepSeek reasoner 的多轮对话

DeepSeek reasoner 模型要求 assistant 消息必须包含 `reasoning_content` 字段，但 langchain-deepseek 没有正确传递：

```python
from langchain_deepseek import ChatDeepSeek

do4 = '''
messages = self._convert_input(input_).to_messages()
reasoning_content_map = {}
for i, msg in enumerate(messages):
    if hasattr(msg, "additional_kwargs"):
        rc = msg.additional_kwargs.get("reasoning_content")
        if rc is not None:
            reasoning_content_map[i] = rc

for i, message in enumerate(payload["messages"]):
    if message["role"] == "assistant":
        if i in reasoning_content_map:
            message["reasoning_content"] = reasoning_content_map[i]
        elif "deepseek-reasoner" in self.model_name and "reasoning_content" not in message:
            message["reasoning_content"] = " "
'''

(
    when(
        ChatDeepSeek._get_request_payload,
        'for message in payload["messages"]:',
    )
    .do(do4)
)
```

## 使用建议

- 补丁代码集中放在一个文件中（如 `utils/dowhen_man.py`），在应用启动时 import
- 补丁中添加注释说明：为什么需要这个补丁、对应的上游 issue
- 当上游修复后，及时移除补丁
- `condition` 参数可以精确控制触发条件，避免影响正常流程

[^dowhen]: [dowhen](https://github.com/lht/dowhen)，2026-07-28 读。