企业官网建设流程全解析

1. 为什么“长跑型” Agent 必须围绕失败来设计

DeerFlow 2.0 是一个开源的 long-horizon SuperAgent Harness，简单说，它能让一个 Agent 连续跑几分钟到几小时，去完成调研、写代码、生成报告这类长任务。它适合谁？适合已经写过单轮 Agent Demo、想进一步复刻“能长跑、跑不死”的 Agent 系统的开发者。它最核心的能力不是让模型多转几圈，而是把模型放进一个能容忍失败的执行环境里。

我先把结论放前面：普通 Agent 和长跑型 Agent 的分界线，不是模型强弱，而是失败恢复能力。一个 Demo 级 Agent 的假设是“一次执行很快结束”，所以它只关心成功路径；而一个真正要跑几十分钟甚至几小时的 Agent，必须假设“中途一定会失败”——网络会抖、模型会抽风、进程会被 kill、租约会过期、子代理会卡住、用户会中途取消。

DeerFlow 2.0 的设计哲学可以压缩成一句话：长任务的本质不是“跑得快”，而是“跑不死”。它建在 LangGraph 之上，用create_agent返回标准的CompiledStateGraph，把推理循环、状态图、检查点、流式、human-in-the-loop 全部交给 LangGraph，自己只专注叠加 Harness 层的能力：Sub-Agent、Run、Worker、RunStore、Memory、Skills、Sandbox、Guardrail、容错。

这篇文章我会拆三块：第一，DeerFlow 的 Harness 分层和 LangGraph 编排骨架；第二，Run / Worker / RunStore 这套“把长任务当一等公民”的设计；第三，给你一份可复制的 LangGraph 节点配置骨架，并用 TaoToken 统一 Key/API 通道接入模型调用，最后跑一次多轮任务验证。

2. TaoToken 前置：统一 Key 与 API 通道

在复刻长跑型 Agent 之前，先把模型调用通道理顺。DeerFlow 这类 Harness 会频繁调用模型，如果每个子代理、每个 Worker 都各自维护一套 Key 和 Base URL，排障会非常痛苦。我的做法是用 TaoToken 统一 Key/API 通道，让 Harness 里所有模型调用都走同一个入口。

TaoToken 官网入口：https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=

API 地址（不加 UTM）：https://taotoken.net/api

你需要先拿到一个 API Key，然后把它作为环境变量注入到 Harness 进程里。这样无论是主 Agent 还是 Sub-Agent，读的都是同一个OPENAI_API_KEY和OPENAI_BASE_URL，切换模型时只改配置，不动代码。

# 在 Harness 运行环境里注入统一通道 export OPENAI_API_KEY="sk-你的TaoTokenKey" export OPENAI_BASE_URL="https://taotoken.net/api"

注意：Base URL 用https://taotoken.net/api，不要在后面拼多余的路径，具体模型名以控制台可用列表为准。

如果你要长期跑编码类 Agent，建议单独看一下 Coding Plan，它更适合高频、长时的编码任务；如果只是验证模型连通性，用模型对话页面点几下就能确认。Key 管理在 API Keys 页面，接入细节在接入文档里。

3. 可复制的 LangGraph 节点配置骨架

下面这份骨架是我按 DeerFlow 的 Harness 思路抽出来的最小可跑版本。它不追求功能全，而是把“长跑”需要的几个关键节点先立起来：状态定义、模型节点、工具节点、检查点、以及一个带租约语义的 Run 记录。

先定义状态。长任务的状态必须显式持久化，不能只放在内存里：

from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] run_id: str owner_worker_id: str | None lease_expires_at: str | None ownership_lost: bool stop_reason: str | None

然后是模型节点。这里用create_agent组装，和 DeerFlow 的做法一致，把推理循环交给 LangGraph：

import os from langchain.agents import create_agent from langchain_openai import ChatOpenAI def build_lead_agent(tools): model = ChatOpenAI( model="gpt-4o-mini", # 以控制台可用模型为准 api_key=os.environ["OPENAI_API_KEY"], base_url=os.environ["OPENAI_BASE_URL"], temperature=0, ) graph = create_agent( model=model, tools=tools, state_schema=AgentState, ) return graph

接着是检查点。长跑型 Agent 的“后悔药”就在这里，用户取消或进程崩溃后要能回滚：

from langgraph.checkpoint.memory import InMemorySaver checkpointer = InMemorySaver() graph = build_lead_agent(tools).compile(checkpointer=checkpointer)

生产环境不要用InMemorySaver，换成持久化后端。DeerFlow 在backend/langgraph.json里把 checkpointer 指向async_provider.py:make_checkpointer，思路就是“检查点必须可持久化、可恢复”。

最后是 Run 记录。这是 DeerFlow 最值钱的设计，把一次执行建模成可租赁、可恢复的对象：

from dataclasses import dataclass, field from datetime import datetime, timedelta @dataclass class RunRecord: run_id: str owner_worker_id: str | None = None lease_expires_at: str | None = None ownership_lost: bool = False status: str = "pending" # pending/running/interrupted def renew_lease(self, worker_id: str, lease_seconds: int = 30): self.owner_worker_id = worker_id self.lease_expires_at = ( datetime.utcnow() + timedelta(seconds=lease_seconds) ).isoformat()

owner_worker_id、lease_expires_at、ownership_lost这三个字段，构成了基于租约的所有权模型。Worker 每隔lease_seconds/3续一次租约，一旦进程死亡，租约过期，其他 Worker 就能接管这个孤儿 Run。

4. 验证请求：跑一次多轮任务

配置好之后，跑一次多轮任务来验证。我构造一个需要多步执行的任务，观察检查点是否生效、Run 状态是否正确流转。

import uuid from langchain_core.messages import HumanMessage run_id = str(uuid.uuid4()) config = {"configurable": {"thread_id": run_id}} record = RunRecord(run_id=run_id) record.renew_lease(worker_id="worker-1") result = graph.invoke( { "messages": [HumanMessage(content="分三步调研 LangGraph 的检查点机制，每步输出一句结论")], "run_id": run_id, "owner_worker_id": record.owner_worker_id, "lease_expires_at": record.lease_expires_at, "ownership_lost": False, "stop_reason": None, }, config=config, ) print("run_id:", run_id) print("status:", record.status) print("stop_reason:", result.get("stop_reason")) print("messages:", len(result["messages"]))

成功时你会看到：run_id稳定输出，messages数量随多轮递增，stop_reason为None。如果中途触发循环检测或预算护栏，stop_reason会变成loop_capped或token_capped，而不是无限空转。

再验证一次恢复能力。用同一个thread_id重新 invoke，检查点会从上次状态继续：

resumed = graph.invoke( {"messages": [HumanMessage(content="继续上一步，补充第四步结论")]}, config=config, ) print("resumed messages:", len(resumed["messages"]))

如果resumed的消息数在上次基础上继续增长，说明检查点持久化生效，这就是长跑型 Agent 的“断点续跑”能力。

5. 本篇常见错排查

报错一：OPENAI_BASE_URL拼错导致 404。常见写法是https://taotoken.net/api/v1，多拼了/v1。正确写法是https://taotoken.net/api，模型名以控制台为准。

报错二：create_agent返回的不是 CompiledStateGraph。检查 LangChain 版本，create_agent是新一代 Agent 原语，不是langgraph.prebuilt的create_react_agent，两者签名不同。

报错三：检查点没生效，每次 invoke 都从头开始。大概率是thread_id每次变了，或者 compile 时没传checkpointer。thread_id必须稳定，它对应一个 Run 的谱系。

报错四：Run 状态卡在running不流转。检查心跳循环是否在跑。租约续期间隔是lease_seconds/3，如果心跳线程被阻塞，租约会过期，其他 Worker 会误判为孤儿并接管。

报错五：子代理把父 Agent 卡死。DeerFlow 的做法是让子代理跑在独立的 asyncio 事件循环上。如果你在同一个循环里同步等待子代理，父 Agent 会被阻塞。用run_on_isolated_subagent_loop这类隔离手段。

报错六：记忆无限膨胀。不要什么都记。DeerMem 的_fact_scope_gate_reason只保留“关于用户、持久、描述性”的事实，自动丢弃临时信息和结论性信息。记忆的核心难题是“记什么”，不是“怎么存”。

6. 下一步：把容错机制按需引入

复刻长跑型 Agent，最忌讳一上来就把 DeerFlow 全套组件搬过来。如果你的项目只有一个 Agent、三个 Tool、跑一次就结束，你只需要 Agent + Tool Registry + Checkpointer。等问题真的出现——进程会死、任务会卡、用户会取消、要并行子任务、记忆会膨胀——再逐步增加 Run/Worker、RunStore、租约与孤儿恢复、Sub-Agent、记忆门控。

验证模型连通性可以去模型对话页面；长期编码或 Agent 任务建议看 Coding Plan；Key 和接入细节分别在 API Keys 与接入文档。把模型调用通道统一到 TaoToken 之后，你就能把精力集中在 Harness 的容错设计上，而不是浪费在到处维护 Key 上。