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

每个模型最多接受 100 万 token 的输入,因此一份完整合同、一页长政策或一小时的智能体历史都能在一次调用中容纳,无需分块。它们还是多模态的: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,包括图像,并且为每个决策添加一句话理由。当智能体根据某个标签来路由工单或停止运行时,这个理由会告诉你原因,从而让智能体更容易调试。
总结消息历史
长时间运行的智能体需要摘要,因为其历史记录最终会超出上下文窗口。一个已经工作了一小时的编码智能体,其历史记录中有数十次文件读取和工具输出,其中大部分它已不再需要逐字保留。每当历史记录填满时,用另一次前沿模型调用来摘要它们会增加成本和延迟。

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,提取会在每一轮重新运行,因此随着客户补充细节,记录会逐渐填充完整。对于 "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 添加到工作区的 提供商密钥 中。然后将 langchain-scaledown 指向该 gateway,并使用你的 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