1. 长周期任务里,普通 Agent 为什么总在第三步就崩了
如果你用 LangChain 的create_agent跑过稍微复杂点的任务,大概率遇到过这种场景:让它调研一个技术方向并输出报告,前两轮对话还挺像样,到第三轮开始它就忘了最初的目标,工具调用开始重复,上下文越堆越长,最后要么超 token 报错,要么给你一份前后矛盾的东西。这不是模型不行,是普通 Agent 的架构本身就不为长周期任务设计。
普通 Agent 的工作模式是「你问一句、它做一步」,本质是个临时工。它没有显式的任务清单,没有把大上下文卸载到外部存储的能力,也没有把子任务隔离到独立上下文里的机制。一旦任务超过五六步,状态就开始漂移。而 Deep Agents 要解决的就是这个问题——它是 LangChain 官方推出的生产级 Agent Harness(智能体驾驭层),基于 LangGraph 做执行引擎、LangChain 做工具与模型层,面向长周期、多步骤、复杂目标设计。一句话对比:普通 Agent 是临时工,Deep Agents 是项目负责人,你给一个目标,它拆计划、分工、执行、沉淀、交付。
这篇文章面向的是已经在用 LangChain/LangGraph 搭 Agent、但被长任务稳定性折磨过的开发者。我会给出可复制的 Harness 配置骨架、状态与工具编排示例,并且把模型 endpoint 切到 TaoToken 之后做一次完整的连通性验证。你跟着做,能拿到一个能跑小时级任务、崩溃后能断点续跑的 Agent 骨架。核心检索词先摆在这:Deep Agents 是基于 LangGraph+LangChain 的生产级 Agent Harness,适合长周期多步骤复杂目标场景。
先说清楚它和普通 Agent 的能力差距在哪。普通create_agent工具调用和多轮对话都有,但任务规划是隐式的、容易混乱;文件系统要自己实现;子 Agent 要自己搭;长期记忆很有限;沙箱执行基本没有;生产就绪需要大量二次开发。Deep Agents 把这些全部内置:显式待办列表可追踪、内置虚拟文件系统做上下文卸载、内置子 Agent 做上下文隔离、跨会话持久记忆、安全沙箱执行、开箱即用的生产运行时。这个差距不是「好用一点」,是「能不能跑完」的区别。
我试过用一个普通 Agent 跑「调研三个竞品并生成对比报告」,到第二个竞品时它就开始把第一个竞品的结论混进来。换成 Deep Agents 的规划+子 Agent 模式后,每个竞品由独立子 Agent 处理,主 Agent 只做编排和汇总,上下文不互相污染,任务能完整跑完。这就是 Harness 层的价值——它把 LLM 的「智能」包装成可观测、可控制、可稳定运行的复杂任务执行环境。
2. TaoToken 前置:把模型 endpoint 统一到可切换的入口
在动手写 Harness 之前,先把模型接入这层理清楚。Deep Agents 本身是模型无关的,支持所有工具调用型 LLM,这意味着你可以把底层模型换成任何兼容 OpenAI 接口的服务。生产环境里我建议把模型 endpoint 统一到一个可切换的入口,这样换模型、做灰度、控成本都不用改业务代码。TaoToken 就是这样一个入口,官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 地址是 https://taotoken.net/api 。
为什么要在 Harness 之前先做这层?因为长周期任务的 token 消耗是普通对话的几十倍,多子 Agent 并行时更夸张。如果模型 endpoint 写死在代码里,你想换个更便宜的模型跑调研子 Agent、用更强的模型跑汇总,就得改一堆地方。统一入口之后,你只需要在配置里改 Base URL 和 Model ID,Harness 骨架完全不用动。
具体要准备三样东西,这也是后面所有配置的基础三件套:Base URL、API Key、Model ID。Base URL 填https://taotoken.net/api,注意这里不加任何 UTM 参数,UTM 只用于官网跳转归因。API Key 在控制台创建,地址是 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。Model ID 根据你选的模型填,比如gpt-4o、claude-sonnet-4-5这类工具调用能力强的模型。
这里有个容易踩的坑:很多人把 Base URL 写成https://taotoken.net/api/v1,然后在 LangChain 里又配了一次/v1,结果路径变成/api/v1/v1/chat/completions,直接 404。LangChain 的ChatOpenAI默认会在 Base URL 后面拼/chat/completions,所以 Base URL 填到/api就够了。如果你用的是 OpenAI 官方 SDK,它默认拼/v1/chat/completions,那 Base URL 要填https://taotoken.net/api/v1。这个差异取决于你用的客户端,配之前先确认清楚。
另外,长周期任务建议在控制台里给 Key 设置好额度上限和用量告警。多子 Agent 并行时 token 消耗曲线很陡,没有告警的话容易在半夜跑爆。TaoToken 控制台支持按 Key 维度看用量,这个对排查「哪个子 Agent 在烧 token」很有用。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,里面有各语言客户端的完整示例,配之前扫一眼能省不少调试时间。
如果你后面要跑长期编码或 Agent 类任务,可以考虑 Coding Plan,地址是 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。它更适合高频、长周期的编码场景,和这篇讲的 Harness 搭配起来用比较顺。但如果你只是想先验证模型通不通,用按量计费的 Key 就够了,不用一上来就上套餐。
3. 可复制的 Harness 配置骨架与状态编排
这一节是全文的核心,给出可以直接抄的配置。先装依赖,Deep Agents 的包名和 LangChain 生态保持一致:
pip install langchain-deepagents langchain-openai langgraph然后是模型配置。我建议把模型配置单独抽成一个模块,方便后面切换。下面这段是config.py,注意 Base URL 和 Model ID 的写法:
# config.py import os from langchain_openai import ChatOpenAI # 从环境变量读取,不要硬编码 Key TAOTOKEN_API_KEY = os.environ.get("TAOTOKEN_API_KEY") TAOTOKEN_BASE_URL = "https://taotoken.net/api" def build_llm(model_id: str = "gpt-4o", temperature: float = 0.2): """构建统一入口的 LLM 实例,所有子 Agent 共用这个工厂函数""" return ChatOpenAI( model=model_id, api_key=TAOTOKEN_API_KEY, base_url=TAOTOKEN_BASE_URL, temperature=temperature, timeout=120, # 长任务给足超时 max_retries=3, # 网络抖动自动重试 )这里base_url填的是https://taotoken.net/api,没有/v1。timeout给到 120 秒是因为长周期任务里单次模型调用可能比较慢,默认超时容易误杀。max_retries=3是必须的,多子 Agent 并行时偶发的网络抖动很常见,没有重试会导致整个任务失败。
接下来是 Harness 骨架。Deep Agents 的核心是create_deep_agent工厂函数,一行生成完整 Agent,内置系统提示词模板规范思考、行动、输出格式。下面这段是harness.py:
# harness.py from langchain_deepagents import create_deep_agent from config import build_llm # 主模型负责编排和汇总,用强一点的 main_llm = build_llm(model_id="gpt-4o", temperature=0.2) # 子 Agent 可以按需用不同模型,这里先共用 agent = create_deep_agent( llm=main_llm, system_prompt=( "你是一个生产级任务编排 Agent。" "收到复杂目标后,先用 write_todos 拆解为可执行待办," "每完成一步更新状态,失败时调整计划而非盲目重试。" "长文本中间产物写入虚拟文件系统,不要全部留在上下文里。" ), tools=[], # 自定义工具在这里注册 subagents=["researcher", "coder", "writer", "reviewer"], )subagents列表里的名字对应内置的子 Agent 类型,每个子 Agent 有独立上下文,主 Agent 只做编排和汇总。这是避免上下文污染的关键——如果让主 Agent 自己干所有活,跑到第五步上下文就爆了。
状态编排这块,Deep Agents 内置了write_todos工具,待办状态分pending、in_progress、completed三种。你不需要自己实现状态机,但需要在系统提示词里明确要求它「每步审查 + 更新」。下面是一个更完整的配置,带上了虚拟文件系统和持久化:
# harness_full.py from langchain_deepagents import create_deep_agent from langgraph.checkpoint.postgres import PostgresSaver from config import build_llm main_llm = build_llm(model_id="gpt-4o") # 持久化:崩溃/部署后断点续跑 checkpointer = PostgresSaver.from_conn_string( "postgresql://user:pass@localhost:5432/agent_state" ) agent = create_deep_agent( llm=main_llm, system_prompt=( "你是项目负责人型 Agent。" "1) 收到目标先 write_todos 拆解;" "2) 大段资料用 write_file 存到 VFS,上下文只留摘要;" "3) 子任务分派给 subagent,主上下文只保留结论;" "4) 每步完成后更新 todo 状态。" ), tools=[], subagents=["researcher", "coder", "writer", "reviewer"], checkpointer=checkpointer, # 断点续跑 )checkpointer是 LangGraph 提供的持久化机制,配上 PostgreSQL 之后,任务跑到一半进程崩了,重启后能从最后一个 checkpoint 继续,不用从头再来。这对小时级任务是刚需。如果你暂时不想上 PostgreSQL,可以先不传checkpointer,但生产环境一定要加。
工具注册这块,tools列表里放的是自定义工具。MCP 协议接入的外部工具、Shell 执行、数据库查询都从这里进。注意不要在这里直连生产库,用只读账号或者沙箱环境。下面是一个自定义工具的示例:
from langchain_core.tools import tool @tool def search_docs(query: str) -> str: """搜索内部文档库,返回相关片段""" # 实际实现接你的检索服务 return f"关于 {query} 的检索结果..." agent = create_deep_agent( llm=main_llm, tools=[search_docs], subagents=["researcher", "writer"], )配置写完后,运行任务的方式是:
result = agent.invoke( {"messages": [{"role": "user", "content": "调研 Deep Agents 的三个核心能力并写一份 1500 字报告"}]}, config={"configurable": {"thread_id": "task-001"}}, # 持久化需要 thread_id ) print(result["messages"][-1].content)thread_id是持久化的关键,同一个 thread_id 重启后能续跑。不同任务用不同 thread_id,避免状态串台。
4. 验证请求:确认 Harness 真的连上了 TaoToken
配置写完不能直接跑大任务,先做一次最小连通性验证。这一步的目的是确认三件事:Base URL 拼对了、API Key 有效、模型返回的是标准 OpenAI 格式。很多人跳过这步直接跑 Harness,结果报错信息被 LangGraph 的状态机吞掉,排查半天。
先写一个最小验证脚本verify.py:
# verify.py from config import build_llm llm = build_llm(model_id="gpt-4o") resp = llm.invoke("只回复两个字:连通") print("返回内容:", resp.content) print("模型标识:", resp.response_metadata.get("model_name")) print("token 用量:", resp.response_metadata.get("token_usage"))运行python verify.py,预期输出类似:
返回内容: 连通 模型标识: gpt-4o token 用量: {'prompt_tokens': 12, 'completion_tokens': 2, 'total_tokens': 14}看到返回内容: 连通就说明 Base URL、Key、Model ID 三件套都对了。如果token_usage里有数字,说明请求确实打到了 TaoToken 并正常计费。这一步过了,再跑 Harness。
接着验证 Harness 层。跑一个带规划的小任务,观察它是否真的调用了write_todos:
# verify_harness.py from harness import agent result = agent.invoke( {"messages": [{"role": "user", "content": "把「写一篇技术博客」拆成待办列表,先不要执行"}]}, config={"configurable": {"thread_id": "verify-001"}}, ) for msg in result["messages"]: print(f"[{msg.type}] {msg.content[:200]}")预期能看到 Agent 调用write_todos工具,输出类似:
[human] 把「写一篇技术博客」拆成待办列表,先不要执行 [ai] 我来拆解这个任务... [tool] todos 已更新: [pending] 调研主题, [pending] 列大纲, [pending] 写初稿... [ai] 已拆解为 5 个待办,当前状态均为 pending看到[tool]那行说明 Harness 的规划能力生效了。如果只看到[ai]没有[tool],说明模型没有触发工具调用,检查两点:一是模型是否支持工具调用(有些轻量模型不支持),二是系统提示词里有没有明确要求「先 write_todos」。
再验证子 Agent 隔离。跑一个需要分工的任务:
result = agent.invoke( {"messages": [{"role": "user", "content": "调研 LangGraph 和 LangChain 的关系,用 researcher 子 Agent 做"}]}, config={"configurable": {"thread_id": "verify-002"}}, )预期能看到主 Agent 调用task工具生成子 Agent,子 Agent 的上下文和主上下文是分开的。这一步验证通过,说明你的 Harness 骨架完整可用。
最后验证持久化。如果你配了checkpointer,跑一个任务到一半 Ctrl+C 中断,然后用同一个thread_id重新 invoke,看它是否从断点继续。这个验证比较麻烦,但生产环境必须做一次,否则真出问题时你不知道能不能恢复。
5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth
这一节按真实报错来。下面这几个是我在配 Deep Agents + TaoToken 时实际遇到过的,按报错信息对照排查。
报错一:401 Unauthorized
openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}原因通常是 Key 没读到或者读错了。检查os.environ.get("TAOTOKEN_API_KEY")是否返回 None。如果你在 shell 里export了但 IDE 没继承,重启 IDE。另外确认 Key 没有多余空格,从控制台复制时容易带上换行。如果 Key 是对的还报 401,检查是不是把官网地址当成了 API 地址——官网是https://taotoken.net,API 是https://taotoken.net/api,两者不能混。
报错二:local proxy failed / Connection error
openai.APIConnectionError: Connection error. httpx.ConnectError: [Errno 111] Connection refused这个报错信息里带「proxy」字样时,先检查你的环境变量里有没有HTTP_PROXY、HTTPS_PROXY、ALL_PROXY。有些开发机全局配了代理,LangChain 的 httpx 客户端会自动读取,导致请求走错路径。临时清掉:
unset HTTP_PROXY HTTPS_PROXY ALL_PROXY然后重跑验证脚本。如果清掉后正常,说明是环境变量污染,不是代码问题。注意这里说的是清理本机环境变量,不是让你去配什么网络工具,两者完全不同。
报错三:reading choices / KeyError 'choices'
KeyError: 'choices' # 或 TypeError: 'NoneType' object is not subscriptable (reading 'choices')这个报错说明返回的 JSON 里没有choices字段,通常是 Base URL 拼错了,请求打到了一个不返回 OpenAI 格式的地址。检查你的base_url是不是写成了https://taotoken.net/api/v1而客户端又拼了一次/v1。正确写法是ChatOpenAI配https://taotoken.net/api,OpenAI 官方 SDK 配https://taotoken.net/api/v1。打印一下实际请求 URL 确认:
print(llm.openai_api_base) # 确认实际 base_url报错四:OAuth / token 过期类
Error: OAuth token expired, please re-authenticate如果你用的是某些 CLI 工具(比如 Claude Code 类)接入,可能会遇到 OAuth 相关的报错。这类工具通常有自己的认证流程,和 API Key 是两套机制。排查思路是:确认你用的是 API Key 模式而不是 OAuth 模式。在 Claude Code 场景下,需要配置ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY两个环境变量,Base URL 指向https://taotoken.net/api,Key 用控制台创建的。如果之前配过 OAuth,先清掉旧的凭证缓存再重配。
报错五:子 Agent 不触发 / task 工具没被调用
这个不是报错,是「静默失败」。表现是任务跑了但没看到子 Agent 分工。原因通常是系统提示词里没明确要求分派,或者subagents列表为空。检查create_deep_agent的subagents参数有没有传,以及系统提示词里有没有「子任务分派给 subagent」这类指令。模型不会自动猜你要用子 Agent,得明确告诉它。
报错六:断点续跑失效
配了checkpointer但重启后还是从头跑。检查两点:一是thread_id是否一致,不同 thread_id 就是不同任务;二是 PostgreSQL 连接是否正常,checkpointer 写不进去的话不会报错但也不生效。可以查一下数据库里有没有 checkpoint 记录:
SELECT thread_id, checkpoint_id FROM checkpoints ORDER BY checkpoint_id DESC LIMIT 5;有记录说明持久化生效了,没有就是连接或配置问题。
排查完这些,你的 Harness 基本就稳了。如果还有拿不准的,接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API Key 在 https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 管理。想先验证模型对话效果,可以用 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 直接试。
6. 把 Harness 跑进生产:从验证到长期任务的落地路径
验证通过之后,下一步是把它跑进真实场景。这里给一条从验证到生产的落地路径,按顺序做,不要跳步。
第一步,把验证脚本里的thread_id换成业务维度的标识,比如task-{user_id}-{timestamp}。这样每个用户的任务状态隔离,排查问题时能按 thread_id 定位。第二步,给不同子 Agent 配不同模型。调研类子 Agent 用便宜快速的模型,汇总和写作类用强模型,成本能降一大截。在create_deep_agent里可以通过子 Agent 配置分别指定,不用所有子 Agent 共用一个 LLM。
第三步,加人机闭环。关键操作(比如写数据库、发邮件、部署)前插入人工审批节点。LangGraph 支持 interrupt,在工具调用前暂停等人工确认。这个对生产环境是必须的,不然 Agent 跑飞了没人拦得住。第四步,接可观测性。LangGraph 自带 tracing,把日志接到你的监控系统,重点看三个指标:单任务 token 消耗、子 Agent 调用次数、断点续跑触发次数。这三个指标异常说明 Harness 配置有问题。
第五步,做成本控制。在 TaoToken 控制台给 Key 设额度上限,同时在 Harness 里加一个 token 预算检查,超过阈值就暂停任务等人工介入。长周期任务最容易失控的就是成本,多子 Agent 并行时 token 消耗是线性叠加的。
关于模型选择,Deep Agents 是模型无关的,但生产环境建议选工具调用能力强的模型。工具调用弱的模型会频繁出现「该调工具时不调、不该调时乱调」的问题,在长任务里会被放大。你可以先用模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 对比几个候选模型的工具调用表现,再决定用哪个跑主 Agent。
最后说一个实际经验:Harness 的稳定性不取决于模型多强,取决于状态管理做得多细。我见过用顶级模型但没配 checkpointer 的 Harness,跑两小时崩一次从头再来;也见过用中等模型但状态管理扎实的,能稳定跑一整天。Deep Agents 把状态管理的骨架给你了,但checkpointer、thread_id、子 Agent 隔离这些细节要自己配到位。配好之后,长周期任务才真的能「跑完」,而不是「跑到一半」。
如果你要跑的是长期编码或 Agent 类高频任务,Coding Plan 在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,和这套 Harness 搭配用比较合适。先把这篇的验证脚本跑通,再上套餐,顺序别反。