LangChain 发布 langchain-scaledown 集成,用 ScaleDown 约束解码模型管理长时智能体上下文
Building Long-Running Agents with ScaleDown and Deep Agents
LangChain 发布 langchain-scaledown 集成,并在 LangSmith LLM Gateway 中支持 ScaleDown 的约束解码模型(CDM),用于降低长时智能体上下文管理的成本和延迟。
文章给出 ScaleDown 约束解码模型与 LangChain、Deep Agents 中间件的具体配置方法,读者可以按场景迁移用于压缩智能体上下文。
构建长时间运行的 agent 归根结底是上下文管理的艺术:决定哪些信息进入上下文窗口,以及何时进入。这意味着要总结旧历史、压缩大型工具结果,并从长输入中提取正确的数据。每一步都是一个决策。过去我们用 LLM 来做这些决策,但在长时间运行的 agent 所需的规模下,这既慢又贵。
为了让这些决策更快、更便宜,我们现在有了 langchain-scaledown 集成,以及 LangSmith LLM Gateway 中的支持,用于 ScaleDown 的约束解码模型(CDM)。通用 LLM 一次一个 token 地写出任意字符串,而 CDM 针对单一任务训练,只返回该任务允许的答案,比如你列表中的一个标签,或输入中的片段。这让它可以小得多、快得多、便宜得多:ScaleDown 报告称,在这些狭窄任务上,其模型比前沿模型快 200 倍、便宜 467 倍。
该集成附带三个用于 LangChain 和 Deep Agents 的 中间件:一个总结旧历史,一个压缩大型工具结果,一个从对话中提取结构化数据。在这篇文章中,我们将介绍 ScaleDown 的模型是什么,以及如何使用它们来缩减你的 agent 上下文。
ScaleDown 提供哪些模型?
ScaleDown 有四个模型,每个都针对单一任务训练,因此每个都能比通用 LLM 高效得多:
- Summarize 为长文本写出简短摘要。
- Compress 将一块上下文裁剪到与问题相关的部分。
- Extract 从文本中提取你描述的字段,每个字段都带有置信度分数及其在源文本中的位置。
- Classify 根据你定义的标签对文本打分,并为每个标签返回一个概率。你可以决定是否返回做出该分类的自然语言理由。

每个模型最多接受 100 万 token 的输入,因此一份完整合同、一页长政策,或一小时的 agent 历史都能在一次调用中容纳,无需分块。它们还是多模态的:Extract、Summarize 和 Classify 除了非结构化文本外,还能直接接受图像和扫描版 PDF,因此截图或扫描发票也可以作为输入。
在 ScaleDown 自己的基准测试中,以 GPT-5.5、GPT-5.4 Mini 和 GPT-5.4 Nano 在公开数据集上的平均值为对照,Summarize 便宜 41 倍、快 2.7 倍,Compress 便宜 138 倍、快 1.8 倍,同时两者在准确率上都得分更高。
如果你用过像 Jev 这样的决策模型,ScaleDown Classify 会让你感到熟悉:两者都返回带概率的类型化答案。Jev 每个问题只接受文本,最多 32K token 的状态。Classify 最多接受 100 万 token,包括图像,并且为每个决策添加一句话理由。当 agent 根据标签路由工单或停止运行时,理由会告诉你原因,这让 agent 更容易调试。
总结消息历史
长时间运行的智能体需要摘要,因为其历史记录最终会超出上下文窗口。一个已经工作了一小时的编码智能体,其历史记录中有几十次文件读取和工具输出,其中大部分它已不再需要逐字保留。每次历史记录填满时,用另一次前沿模型调用来摘要它们会增加成本和延迟。

ScaledownSummarizationMiddleware 替换了 Deep Agents 内置的摘要中间件。它接受相同的 trigger 和 keep 选项来决定何时摘要以及保留哪些最近的消息,并由 ScaleDown 写入摘要:
from deepagents import create_deep_agent
from langchain_scaledown import ScaledownSummarizationMiddleware
agent = create_deep_agent(
model="openai:gpt-6-sol",
tools=[run_tests],
middleware=[
ScaledownSummarizationMiddleware(
"openai:gpt-6-sol", # the agent’s model, used for its context window
trigger=("fraction", 0.8),
keep=("fraction", 0.1),
),
],
)智能体保留自己的模型来规划和编写代码。只有摘要会转移到 ScaleDown,默认情况下它返回与 LangChain 的摘要提示相同的部分(意图、摘要、产物和后续步骤)。如果 ScaleDown 调用失败,中间件会保持历史记录不变,智能体继续运行。同一个中间件也适用于 LangChain 的 create_agent。
压缩工具结果
长时间运行的智能体会调用大量工具,而每个大型结果都会在剩余的运行过程中留在上下文中。一个抓取的网页可能长达数万个字符,而智能体只需要其中一段。之后的每一步都要再次为这些 token 付费,而且随着其上下文被填满,智能体保持任务专注的能力也会下降。

ScaledownToolCompressionMiddleware 为你的工具添加了一个可选的 compression_query 参数。当模型只需要结果的一部分时,它在工具调用中填入 compression_query,例如 "refund policy for EU orders"。中间件在工具运行前将其剥离,然后 ScaleDown 在模型看到输出之前将其压缩到相关部分:
from deepagents import create_deep_agent
from langchain_scaledown import ScaledownToolCompressionMiddleware
agent = create_deep_agent(
model="openai:gpt-6-luna",
tools=[search_docs, fetch_url],
middleware=[ScaledownToolCompressionMiddleware(tools=["fetch_url"])],
)你的工具无需更改,因此这适用于任何工具,包括 MCP 工具。在此追踪记录中,ScaleDown 在模型看到之前将一个很长的文档页面削减了近 90%。压缩后的结果以类似这样的说明开头,因此模型如果需要完整页面,可以在不带 compression_query 的情况下再次调用该工具:
[Compressed by ScaleDown for "refund policy for EU orders":
18,402 -> 3,110 chars.
Call this tool again without `compression_query` for the full output.]提取结构化数据
智能体不断从它们读取的内容中提取特定的值,比如一个 ID、一个日期,或下一次工具调用所需的某个字段。这种提取位于智能体循环内部,因此它必须足够快,不会给每一轮增加延迟,并且足够精确,使工具可以直接使用该值。
例如,在支持智能体提交工单之前,它需要订单号、错误消息和客户的电子邮件。主模型可以将这些填入工具参数,但它可能会改写错误代码或打错电子邮件地址,而不是精确复制。
ScaledownExtractionMiddleware 将这一步交给 ScaleDown。你用平实的语言描述每个字段,Extract 会从文本中复制该值。一个字段也可以是一组标签,ScaleDown 用 Classify 来回答,这样工单就可以按问题类型进行路由。在一次运行的第一次模型调用之前,中间件将用户的消息发送给 Extract,并将字段保存到智能体状态中的 scaledown_extraction 下,你的工具可以在那里读取它们:
from deepagents import create_deep_agent
from langchain.tools import ToolRuntime, tool
from langchain_scaledown import ScaledownExtractionMiddleware
ticket = ScaledownExtractionMiddleware(
entities={
"order_id": "Order number the customer mentions",
"error_message": "Exact error message or code the customer saw",
"customer_email": "Customer's email address",
"issue_type": {
"labels": [
{"name": "billing", "rubric": "Is this about a charge, refund, or payment?"},
{"name": "technical", "rubric": "Is this about a bug or something not working?"},
{"name": "account", "rubric": "Is this about login, password, or account settings?"},
]
},
},
context_chars=0, # skip evidence snippets to keep state small
)
@tool
def file_ticket(runtime: ToolRuntime) -> str:
"""File a support ticket with the details gathered so far."""
fields = runtime.state.get("scaledown_extraction", {}).get("fields", {})
return crm.create_ticket(**fields) # your ticketing system
agent = create_deep_agent(model="openai:gpt-6-sol", tools=[file_ticket], middleware=[ticket])Extract 只读取用户的消息,从不读取 agent 的回复,因此每个值都是客户实际说过的话。配合 checkpointer,extraction 会在每一轮重新运行,因此随着客户补充细节,记录会逐渐填充完整。对于 "Order A-1042 failed with ERR_PAYMENT_DECLINED. I already tried a different card.",返回的字段为:
{
"order_id": "A-1042",
"error_message": "ERR_PAYMENT_DECLINED",
"customer_email": None, # not mentioned yet
"issue_type": "billing",
}分类漂移
agent 会根据其当前状态决定下一步,例如任务是否完成,或上一次工具调用是否成功。对状态的分类在每一轮都会发生,因此它需要既便宜又快速。
在长时间运行的 agent 中,值得捕捉的一种状态是漂移:在同一个失败的工具调用上循环,或陷入无产出的循环。没有现成的中间件可用于此,但你可以从自己的 自定义中间件 调用任何 ScaleDown 模型。下面是一个断路器,它在每次模型调用后运行,将历史记录发送给 ScaleDown Classify,并在偏离轨道时停止运行:
import os
import requests
from deepagents import create_deep_agent
from langchain.agents.middleware import after_model
from langchain_core.messages import AIMessage, get_buffer_string
LABELS = [
{"name": "progressing", "rubric": "Is the agent making new progress on the user's task?"},
{"name": "looping", "rubric": "Is the agent repeating the same steps without new results?"},
{"name": "drifting", "rubric": "Is the agent working on something the user did not ask for?"},
]
@after_model(can_jump_to=["end"])
def circuit_breaker(state, runtime):
result = requests.post(
"https://api.scaledown.xyz/classify",
headers={"x-api-key": os.environ["SCALEDOWN_API_KEY"]},
json={"text": get_buffer_string(state["messages"]), "labels": LABELS},
).json()
label = result["top_label"]
if label != "progressing" and result["scores"][label] > 0.8:
return {"jump_to": "end", "messages": [AIMessage(f"Stopped: {result['reasoning']}")]}
return None
agent = create_deep_agent(model="openai:gpt-6-sol", tools=[run_tests], middleware=[circuit_breaker])Classify 会为每个标签返回一个概率,因此你可以选择断路器触发前所需的置信度。它的一行原因会显示在最终消息中,因此你可以看到运行停止的原因。一个更温和的版本不会在第一次触发时就结束运行:它会把原因添加到对话中,作为改变方向的提示,只有在 agent 再次漂移时才停止。
这四个部分展示了 ScaleDown 模型适用的一些场景,但列表并不详尽。你可以混合搭配它们,例如对较长的工具结果运行 extract / summarize,或使用分类进行运行时模型路由。查看 ScaleDown Cookbooks,了解在不同类型数据上使用 CDMs 模型的其他方式。
LangSmith LLM Gateway 中的 ScaleDown
LangSmith LLM Gateway 让你用一个 LangSmith API key 调用跨提供商的模型,追踪每一次调用,并强制执行你的 支出 和 速率限制。ScaleDown 现在是其 直接模型访问提供商 之一,位于 /scaledown 路径。
管理员只需一次将你的 ScaleDown key 作为 SCALEDOWN_API_KEY 添加到工作区的 provider secrets 中。然后将 langchain-scaledown 指向网关,并使用你的 LangSmith key 进行身份验证:
export SCALEDOWN_BASE_URL="https://gateway.smith.langchain.com/scaledown"
export SCALEDOWN_API_KEY="$LANGSMITH_API_KEY"此后,每次 ScaleDown 调用都会显示为一条 LangSmith trace,并与其他模型调用一样计入相同的限制。
何时不使用 ScaleDown
对于开放式生成,例如编写代码、起草回复和规划,请保留通用 LLM。这遵循我们 Jev 文章 中“默认便宜,例外时用前沿模型”的方法:对困难部分使用大型通用 LLM,对周围的高频步骤使用专用模型。
ScaleDown 足够快速且便宜,可以在每一轮运行,因此你的 agent 可以做出上下文管理所依赖的结构化决策,而不会拖慢主循环,也不会为此支付前沿模型的价格。在长时间运行的 agent 中,这些决策几乎在每一轮都会发生,因此节省会迅速累积。
开始使用
安装集成 并 设置你的 ScaleDown API key:
uv pip install -U langchain-scaledown
export SCALEDOWN_API_KEY="..."新的 ScaleDown 账户可获得 5000 万免费 token。ScaleDown 集成文档 涵盖了每个中间件的配置选项。试试 langchain-scaledown,并在 LangChain 论坛、X 或 GitHub 上告诉我们你的想法。
来源:LangChain:Blog · langchain.com