Haystack 3.2.0 实操：给长流程 Agent 加上下文预算，摘要压缩与预算熔断双保险

# 在虚拟环境里升级
pip install -U "haystack-ai>=3.2.0"

# 确认版本到位
python -c "import haystack; print(haystack.__version__)"

升级后先做两项自查，都是 3.2.0 的破坏性变更：其一，Toolset 之间用 + 号组合的写法被彻底移除——如果项目里有 toolset_a + toolset_b，改成把多个 Toolset 放进列表传入；以前用 + 拼过 Toolset 再序列化过的管道，升级后会直接加载失败，必须改写成列表形态重新序列化。其二，序列化产物里带 Jinja custom_filters 的 OutputAdapter 或 ConditionalRouter 组件，现在必须用 Pipeline.load(..., unsafe=True) 才能加载——这是有意的安全收紧，改动只影响加载环节，不影响运行。

# 全库搜这两类旧写法，命中即登记改造清单
# Linux / macOS
grep -rn "toolset_a + toolset_b\|custom_filters" --include="*.py" .
# 改法一：+ 号组合 → 列表
agent = Agent(chat_generator=llm, tools=[toolset_a, toolset_b, search_tool])
# 改法二：带自定义 Jinja 过滤器的序列化管道 → 显式 unsafe 加载
pipe = Pipeline.load("pipeline.yaml", unsafe=True)  # 仅限可信来源的文件

from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack.tools import tool

@tool
def search_orders(query: str) -> str:
    """在订单库里检索与 query 相关的订单记录，返回JSON文本。"""
    # 换成你的真实数据源
    return '[{"order_id": "A1001", "status": "shipped"}]'

@tool
def get_refund_policy(topic: str) -> str:
    """查询售后/退款政策条目。"""
    return "7天内未签收可全额退款，需提供订单号。"

agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"),
    tools=[search_orders, get_refund_policy],
    system_prompt="你是售后助理。逐条核对订单后再下结论，查不到就说明原因。",
)

result = agent.run({"messages": [ChatMessage.from_user(
    "核对这20笔订单的退款资格，逐笔给依据并汇总成清单。")]})
print(len(result["messages"]))  # 观察历史长度随步数增长

给 Agent 派一个天然需要十几二十步的任务（逐笔核对 20 笔订单），然后盯着 result["messages"] 的长度看历史怎么滚雪球：每一步的工具调用、工具结果、中间推理都整段留在历史里。跑到任务后段，单次请求的 token 数可能是开头的好几倍——费用在涨，而模型对开头信息的注意力在稀释。这就是基线病灶，后面两步分别治它。

# 模块路径与参数名以 3.2.0 官方文档为准（下同）
from haystack.components.agents import SummarizationCompactor

compactor = SummarizationCompactor(
    generator=OpenAIChatGenerator(model="gpt-4o-mini"),  # 摘要可用更便宜的模型
    min_keep_steps=3,   # 最近 3 步原样保留，绝不压缩
)

# 挂载方式与参数形态随版本而定：把 compactor 配置进 Agent 的
# 上下文管理入口后重跑 Step 2 的任务，对比压缩前后的 messages

SummarizationCompactor 的工作方式：当运行历史变长时，用一个 LLM 调用把久远的步骤蒸馏成一段摘要文本，替换掉原始的多条消息；min_keep_steps 指定的最近若干步保持原样——这很重要，因为模型对「刚才发生了什么」的精确记忆直接影响下一步决策，摘要只该发生在久远历史上。重跑 Step 2 的任务后对比：任务结论不变形，历史长度显著下降，后段请求的 token 成本随之回落。

调参经验：min_keep_steps 不要设太小——压得太激进，模型会忘记自己刚调过什么工具，开始重复检索；也别太大，压了个寂寞。对逐条核对类任务，保留最近 2-4 步是常见起点，按「压缩后结论是否变形」来微调。摘要模型选便宜档即可，它做的是信息蒸馏而不是决策。

from haystack.components.agents import TokenBudgetHook

budget_hook = TokenBudgetHook(
    budget=120_000,  # 整次运行的 token 预算上限（示例值）
)

# 把 hook 挂进 Agent 运行（挂载入口以官方文档为准），触发后的
# 返回结果里带着：已收集的完整 messages + exit_reason
result = agent.run({"messages": [ChatMessage.from_user("核对这50笔订单……")]})
# 读取停车原因与现场（字段名以文档为准）
print(result.get("exit_reason", "completed"))

TokenBudgetHook 解决的是另一半问题：压缩让 Agent 跑得动，但一个被脏数据逼进循环的 Agent 仍可能无限制地烧下去。这个 hook 持续记账，预算耗尽时在下一次模型调用之前让运行干净停止——不是抛异常崩掉，而是带着已收集的全部消息与 exit_reason 正常返回。这对调度系统极友好：外层代码检查 exit_reason，发现是预算触顶就把现场转人工、或换个提示让它带着已有进展续跑，而不是从零重来。

预算怎么估：跑三遍代表性任务，取「正常完成」的平均消耗乘 1.3 作冗余系数，得出默认预算；再设一个硬顶（比如平均值的 3 倍）作为熔断线。预算线划太紧，正常任务被误杀；划太松，失控任务照样烧钱——用实测数据定线，别拍脑袋。

# 两个机制各管一层，同时挂载（挂载入口以官方文档为准）
# compactor 管单次运行的历史膨胀
# budget_hook 管整次运行的资源上限
#
# 上线前写一个最小回归脚本：
def test_long_task_stops_cleanly():
    result = agent.run({"messages": [ChatMessage.from_user(TASK_50_ORDERS)]})
    msgs = result["messages"]
    # 断言：要么正常完成，要么预算停车——不允许异常崩溃
    assert result.get("exit_reason", "completed") in ("completed", "budget_exhausted")
    # 断言：压缩生效后历史长度低于未压缩基线
    assert len(msgs) < BASELINE_LEN
    # 断言：最近步骤未被压缩（保留完整工具结果）
    assert TOOL_RESULT_MARK in msgs[-3].text

上线前用三个断言把行为锁死：停得下（任何结局都是体面返回）、压得动（历史长度确实低于基线）、压不坏（最近步骤完整保留、最终结论与未压缩版本一致）。回归脚本进 CI，每次升级 haystack-ai 都跑一遍——实验特性的 API 在小版本间仍可能调整，回归脚本是你的哨兵。

from haystack import Pipeline

pipe = Pipeline()
# 3.2.0：一次加多个组件、一次连多条边，且可链式调用
pipe.add_components(retriever, embedder, ranker) \
    .connect_many([(embedder, retriever), (retriever, ranker)])

# 旧写法仍然有效，但新写法在动态装配场景（按配置拼管道）里
# 能把十几次 add_component / connect 调用收拢成两行

这次发布还带了 add_components() 与 connect_many()——支持一次传入多个组件或多条连接，并且链式调用。对「按配置文件动态拼管道」的团队最实用：原来一段十几行的装配循环，现在收拢成两行，出错面也更小。存量代码不用改，旧写法兼容；新代码建议直接用新写法。