---
title: 自定义与扩展
description: DeerFlow 的可插拔架构意味着系统的大多数部分都可以在不 fork 核心的情况下被替换或扩展。本页面列举了扩展点，并解释如何使用每一个。
---
import { Callout, Cards } from "nextra/components";
# 自定义与扩展
DeerFlow
设计为可适配的。你可以通过编写自定义中间件、添加新工具、构建技能包以及通过
config.yaml 的 use: 字段替换任何内置组件来扩展 Agent 行为。
DeerFlow 的可插拔架构意味着系统的大多数部分都可以在不 fork 核心的情况下被替换或扩展。本页面列举了扩展点，并解释如何使用每一个。
## 自定义中间件
中间件是为 Lead Agent 添加行为的主要扩展点。它们包裹每次 LLM 调用，可以在模型调用前后读取和修改 Agent 的状态。
添加自定义中间件：
1. 实现 `langchain.agents.middleware` 中的 `AgentMiddleware` 接口。
2. 在构建 Agent 时通过 `custom_middlewares` 参数传入你的中间件。
```python
from langchain.agents.middleware import AgentMiddleware
from deerflow.agents.thread_state import ThreadState
class AuditMiddleware(AgentMiddleware):
async def on_start(self, state: ThreadState, config):
# 在每次模型调用前运行
print(f"[审计] 轮次开始：上下文中有 {len(state.messages)} 条消息")
return state, config
async def on_end(self, state: ThreadState, config):
# 在每次模型调用后运行
print(f"[审计] 轮次结束：最后一条消息类型 = {state.messages[-1].type}")
return state, config
```
自定义中间件在链末尾 `ClarificationMiddleware` 之前注入，后者始终最后运行。
## 自定义工具
在 `config.yaml` 的 `tools:` 下注册新工具，将其添加到 Agent：
```yaml
tools:
- use: mypackage.tools:my_custom_tool
api_key: $MY_TOOL_API_KEY
```
你的工具必须是 LangChain `BaseTool` 或用 `@tool` 装饰的函数。它将使用 `use:` 类路径和配置条目中的任何额外字段实例化。
```python
# mypackage/tools.py
from langchain_core.tools import tool
@tool
def my_custom_tool(query: str) -> str:
"""搜索我的自定义数据源。"""
return do_search(query)
```
## 自定义沙箱提供者
沙箱可以通过实现 `SandboxProvider` 接口来替换：
```python
from deerflow.sandbox.sandbox_provider import SandboxProvider
class MyCustomSandboxProvider(SandboxProvider):
def acquire(self, thread_id: str | None = None) -> str:
# 返回 sandbox_id
...
def get(self, sandbox_id: str):
# 返回此 id 的沙箱实例
...
def release(self, sandbox_id: str) -> None:
# 清理
...
```
然后在 `config.yaml` 中引用：
```yaml
sandbox:
use: mypackage.sandbox:MyCustomSandboxProvider
```
## 自定义记忆存储
通过实现 `MemoryStorage` 将基于文件的记忆替换为任何持久化存储：
```python
from deerflow.agents.memory.storage import MemoryStorage
from typing import Any
class RedisMemoryStorage(MemoryStorage):
def load(self, agent_name: str | None = None) -> dict[str, Any]:
...
def reload(self, agent_name: str | None = None) -> dict[str, Any]:
...
def save(self, memory_data: dict[str, Any], agent_name: str | None = None) -> bool:
...
```
在 `config.yaml` 中配置：
```yaml
memory:
storage_class: mypackage.storage:RedisMemoryStorage
```
## 自定义技能
技能是最简单的扩展点。在 `skills/custom/your-skill-name/` 下创建一个目录，并在其中添加 `SKILL.md` 文件。技能将在下次 `load_skills()` 调用时自动被发现。
参见[技能](/docs/harness/skills)页面了解完整的目录结构和 `SKILL.md` 格式。
## 自定义模型
任何与 LangChain 兼容的聊天模型都可以通过在 `use:` 字段中指定来使用：
```yaml
models:
- name: my-custom-model
use: mypackage.models:MyCustomChatModel
# 任何额外字段都作为 kwargs 传递给构造函数
base_url: http://my-model-server:8080
api_key: $MY_MODEL_API_KEY
```
模型类必须实现 LangChain 的 `BaseChatModel` 接口。
## 自定义检查点
线程状态持久化可以使用任何与 LangGraph 兼容的检查点器：
```yaml
checkpointer:
type: sqlite
connection_string: ./my-checkpoints.db
```
对于自定义后端，实现 LangGraph 的 `BaseCheckpointSaver` 接口，并在初始化 `DeerFlowClient` 时以编程方式配置它。