企业官网建设流程全解析

1. 长周期 Agent 的上下文膨胀到底卡在哪

如果你正在做多轮对话的 Agent，大概率遇到过这个场景：第一轮聊得好好的，到第八轮、第十轮，模型开始"忘事"，之前确认过的偏好、已经查过的资料、上一轮定好的方案，它全都不记得了。更糟的是，Token 消耗曲线像坐火箭，每轮请求的输入长度都在涨，成本肉眼可见地失控。

这个问题的根因不在模型本身，而在上下文窗口的管理方式。传统 Agent 的做法是把所有工具调用结果、所有历史消息一股脑塞进 messages 数组，每轮都全量重发。一次网络搜索返回 100KB 文本，一次文件读取返回几千行代码，这些内容全部堆在上下文里，模型在海量信息中逐渐失焦，同时 Token 账单持续攀升。

DeepAgents 是 LangChain 团队开源的一个 Agent 框架，专门针对长周期任务设计。它的核心思路是把"信息流"重新设计：引入文件系统作为上下文缓冲区，大块工具结果自动写入文件，Agent 上下文中只保留路径引用；配合自动摘要和提示缓存机制，显著降低单轮 Token 消耗。它提供三大机制——任务规划（write_todos/read_todos）、文件系统访问（ls/read_file/write_file/edit_file/glob/grep）、子 Agent 委托（task 工具），用 create_deep_agent 创建出来的 Agent 本质是一个编译后的 LangGraph StateGraph，可以直接用 LangGraph 的流式输出、检查点、人机交互等特性。

但 DeepAgents 默认的 StateBackend 只支持单次会话存储，进程一重启，记忆就没了。跨会话记忆需要持久化后端，这正是 Milvus 向量库要解决的问题。本文聚焦 Token 预算机制与 Milvus 跨会话记忆的落地：先拆解上下文膨胀的根因，再给出 Milvus 集合 Schema、嵌入写入与检索召回的可复制配置，最后用多轮会话脚本验证记忆命中率与 Token 消耗变化。适合正在做长周期 Agent、被上下文长度和成本困扰的开发者。

2. 为什么 DeepAgents 需要 Milvus 做跨会话记忆

先说清楚一个概念：DeepAgents 的"记忆"和"上下文"是两回事。上下文是当前这一轮对话窗口里的内容，记忆是跨轮次、跨会话需要保留的信息。前者受模型窗口限制，后者需要外部存储。

DeepAgents 默认的 StateBackend 把文件存在内存里，会话结束就释放。对于单次任务这没问题，但如果你要做一个"记住用户偏好、积累领域知识、维护长期研究进度"的 Agent，内存存储完全不够用。持久化 backend 能解决跨会话数据保留问题，但普通的键值存储又缺少语义检索能力——你没法用"上次讨论过的那个向量库方案"这种模糊描述去精确命中历史记忆。

这就是 Milvus 的切入点。Milvus 是向量数据库，把 Agent 对话内容和重要工具结果转成 embedding 存进去，每次任务执行时通过语义检索找回相关历史记忆。它的计算存储分离架构支持高并发读写，可水平扩展到十亿级向量规模，同时支持高并发查询和流式数据实时更新，适合生产环境的 Agent 记忆存储场景。

在 DeepAgents 里，这个方案通过 CompositeBackend 实现混合存储路由：把不同路径路由到不同 backend。比如/workspace/、/temp/走 StateBackend（临时文件），/memories/、/knowledge/走 StoreBackend + Milvus（持久化数据）。这样 Agent 在写临时工作文件时不落库，只有明确写入/memories/的内容才会持久化并被语义检索。

这里有个关键点：StoreBackend 需要一个 Store 实例，LangGraph 生态里可以用 InMemoryStore 做测试，生产环境换成 Milvus 适配器即可。MilvusStore 的 collection 里存的是 namespace + key + embedding + value 的结构，检索时按 namespace 隔离不同 Agent 或不同用户的记忆空间。

如果你在本地跑 Milvus 有困难，或者想让 Agent 的模型调用走一个稳定的接入层，可以用 TaoToken 这类统一 API 网关。它的 Base URL 是https://taotoken.net/api，兼容 OpenAI 协议，把模型调用和向量存储解耦开，调试时切换模型不用改业务代码。API Key 在控制台生成：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite

3. 可复制配置：Milvus 集合 Schema 与 DeepAgents Backend

这一节给出完整的可复制配置。先装依赖：

pip install deepagents tavily-python langchain-milvus pymilvus langgraph-checkpoint

Milvus 集合的 Schema 设计要围绕"记忆检索"这个目标。核心字段包括：id（主键）、namespace（记忆空间隔离）、key（记忆标识）、embedding（向量）、value（原始文本）、created_at（时间戳）。用 pymilvus 建集合：

from pymilvus import ( connections, Collection, CollectionSchema, FieldSchema, DataType, utility ) connections.connect(host="localhost", port="19530") COLLECTION_NAME = "agent_memories" DIM = 1024 # 与你的 embedding 模型维度一致 if utility.has_collection(COLLECTION_NAME): utility.drop_collection(COLLECTION_NAME) fields = [ FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=64), FieldSchema(name="namespace", dtype=DataType.VARCHAR, max_length=128), FieldSchema(name="key", dtype=DataType.VARCHAR, max_length=256), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=DIM), FieldSchema(name="value", dtype=DataType.VARCHAR, max_length=8192), FieldSchema(name="created_at", dtype=DataType.INT64), ] schema = CollectionSchema(fields, description="DeepAgents cross-session memory") collection = Collection(COLLECTION_NAME, schema) index_params = { "index_type": "HNSW", "metric_type": "COSINE", "params": {"M": 16, "efConstruction": 200}, } collection.create_index(field_name="embedding", index_params=index_params) collection.load() print(f"collection {COLLECTION_NAME} ready, dim={DIM}")

HNSW 索引在召回率和延迟之间平衡较好，M=16、efConstruction=200 是常用起点。metric_type 用 COSINE，因为文本 embedding 通常做归一化后比余弦相似度。

接下来配置 DeepAgents 的 CompositeBackend。这里用 LangGraph 的 InMemoryStore 做演示，生产环境替换成 Milvus 适配的 Store：

from deepagents import create_deep_agent from deepagents.backends import CompositeBackend, StateBackend, StoreBackend from langgraph.store.memory import InMemoryStore # 生产环境替换为 Milvus 适配的 Store 实例 memory_store = InMemoryStore() backend = CompositeBackend( default=StateBackend(), routes={ "/memories/": StoreBackend(store=memory_store), "/knowledge/": StoreBackend(store=memory_store), } )

如果你用 TaoToken 作为模型接入层，在创建 Agent 时指定模型和 Base URL：

import os from tavily import TavilyClient tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"]) def internet_search(query: str, max_results: int = 5) -> str: """执行网络搜索""" results = tavily_client.search(query, max_results=max_results) return " ".join([f"{r['title']}: {r['content']}" for r in results["results"]]) agent = create_deep_agent( tools=[internet_search], system_prompt=( "你是研究专家。将重要发现写入 /memories/ 目录以便跨会话复用。" "写入时用简洁的 key 命名，value 保留关键结论和来源。" ), backend=backend, model="openai:gpt-4o", )

对应的环境变量配置（.env或 shell export）：

export OPENAI_API_KEY="你的 TaoToken API Key" export OPENAI_BASE_URL="https://taotoken.net/api" export TAVILY_API_KEY="你的 Tavily Key"

注意 Base URL 是https://taotoken.net/api，不要加多余路径。模型 ID 按你实际使用的填，比如gpt-4o、claude-sonnet-4-5等。Coding Plan 适合长期编码和 Agent 场景，额度更划算：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite

4. 验证请求：多轮会话脚本与记忆命中率

配置好之后，写一个多轮会话脚本验证记忆是否真的跨会话生效。核心思路：第一轮让 Agent 研究一个主题并写入/memories/，第二轮在新会话里问相关问题，看它能否检索到上一轮的结论。

import uuid from langchain_core.messages import HumanMessage def run_session(agent, user_input, thread_id): config = {"configurable": {"thread_id": thread_id}} result = agent.invoke( {"messages": [HumanMessage(content=user_input)]}, config=config, ) return result["messages"][-1].content # 第一轮：研究并写入记忆 session_1 = str(uuid.uuid4()) out1 = run_session( agent, "研究 Milvus 向量数据库的技术特点，把关键结论写入 /memories/milvus_notes.md", session_1, ) print("=== Session 1 ===") print(out1[:500]) # 第二轮：新会话，问相关问题 session_2 = str(uuid.uuid4()) out2 = run_session( agent, "我之前研究过 Milvus，帮我回忆一下它的核心架构特点是什么？", session_2, ) print("=== Session 2 ===") print(out2[:500])

跑完之后观察两个指标。第一是记忆命中率：第二轮的回答里是否出现了第一轮写入的具体结论（比如"计算存储分离""HNSW 索引"这些关键词）。如果命中，说明 StoreBackend 的检索生效了。第二是 Token 消耗：用 LangSmith 或自己打点记录每轮的usage_metadata，对比"有记忆检索"和"全量历史重发"两种模式的输入 Token 数。

# 打印 Token 消耗 for msg in result["messages"]: if hasattr(msg, "usage_metadata") and msg.usage_metadata: print(f"input_tokens={msg.usage_metadata.get('input_tokens')}, " f"output_tokens={msg.usage_metadata.get('output_tokens')}")

实测下来，在 10 轮以上的长会话里，把大块工具结果写入文件、上下文只留路径引用的做法，输入 Token 能压到全量重发的三分之一到一半。记忆检索本身会带来一点额外开销（embedding 调用 + 向量查询），但相比省下的上下文 Token，整体是划算的。

如果你想单独验证模型对话是否正常，可以用模型对话页面快速测一下：https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite

5. 本篇常见错排查

这一节列几个实际会撞到的报错和排查路径。

401 Unauthorized / invalid api key：最常见。检查OPENAI_API_KEY是否设置正确，Base URL 是否写成https://taotoken.net/api（注意结尾没有斜杠，也没有/v1）。如果你用的是 OpenAI SDK 默认行为，它会自动拼/chat/completions，所以 Base URL 到/api为止。Key 在控制台重新生成一次确认：https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite

local proxy failed / connection refused：如果你本地配了代理环境变量（HTTP_PROXY/HTTPS_PROXY），SDK 会走代理导致连接失败。检查env | grep -i proxy，把相关变量 unset 掉再跑。Milvus 连接失败也会报类似错误，确认localhost:19530端口是否在监听，docker ps看容器状态。

reading 'choices' of undefined：这个报错通常出现在响应体不是标准 OpenAI 格式时。原因可能是 Base URL 配错，请求打到了非兼容端点，或者模型 ID 写错导致服务端返回错误结构。打印原始响应体定位：

import httpx resp = httpx.post( "https://taotoken.net/api/chat/completions", headers={"Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}"}, json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]}, ) print(resp.status_code) print(resp.text[:500])

OAuth / authentication_error：如果你用的是 Claude Code 或 Codex 这类 CLI 工具，它们有自己的认证流程。Codex 的auth.json里需要配置 Base URL、Key、Model ID 三件套。Claude Code 通过环境变量ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY接入。CC Switch 这类工具切换配置时，确认三件套都写全了，缺一个都会认证失败。接入文档在这里：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite

Milvus collection not loaded：建完索引后必须调collection.load()，否则查询会报 collection 未加载。另外插入数据后要collection.flush()才能被检索到。

embedding 维度不匹配：建集合时DIM必须和 embedding 模型输出维度一致。用 1024 维的模型建了 1536 维的集合，插入时会报维度错误。换模型时记得重建集合。

6. 把记忆层接进你的 Agent 工作流

到这里，DeepAgents + Milvus 的跨会话记忆链路已经跑通了。回顾一下关键设计：用 CompositeBackend 做路径路由，临时文件走 StateBackend 不落库，/memories/和/knowledge/走 StoreBackend 持久化；Milvus 集合用 HNSW + COSINE 索引，namespace 字段隔离不同用户或 Agent 的记忆空间；多轮会话脚本验证记忆命中率和 Token 消耗。

实际落地时还有几个可以优化的点。一是记忆写入策略，不要让 Agent 把所有东西都写进/memories/，在 system prompt 里明确"只写关键结论和来源"，避免记忆库膨胀。二是检索 top_k 的调优，默认召回太多会引入噪声，太少会漏掉相关记忆，从 top_k=3 开始试。三是定期清理过期记忆，给 created_at 加 TTL 策略，避免向量库无限增长。

如果你想让 Agent 的模型调用更稳定，把 Base URL 统一指向https://taotoken.net/api，模型切换和额度管理都在一个控制台完成。长期跑 Agent 任务的话，Coding Plan 的额度模型更适合：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite

最后留一个实操建议：先用 InMemoryStore 把整条链路跑通，确认记忆写入和检索逻辑没问题，再换成 Milvus 适配器。这样排查问题时能快速定位是业务逻辑问题还是存储层问题。Milvus 的 docker-compose 起一个单机版就够开发用，生产环境再上集群。