主动推理：让AI智能体按需获取上下文，优化Token成本

这次我们聊的是一个设计思路，而不是某个具体开源模型：怎么让 AI 智能体不再被动接收完整上下文，而是自己判断“我现在缺什么信息”，然后按需去取。

先看问题本质。智能体与模型交互时，token 既是成本单位，也是延迟单位。上下文塞得越满，费用越高、响应越慢，模型反而容易被无关信息带偏；上下文截得太短，关键信息丢失，任务质量直接下滑。传统方案基本是在“全量塞入”和“固定截断”之间取舍，主动推理则换了一条路：把“获取什么上下文”这个动作，交给智能体自己在工作流中决策。

主动推理（Active Inference）在本文语境下并不是脑科学里的抽象理论，而是一种工程策略：智能体先判断当前任务缺什么信息，再决定从对话历史、知识库、数据库、外部工具或文件中获取，最后只把最相关的部分装进上下文。它适合长文档问答、多工具串联、复杂调研、客服辅助等场景。下面会给出可参考的架构、代码模板、测试方法、接口扩展方式和排查清单，你可以直接拿这套思路改造自己的智能体项目。

1. 核心能力速览

先说清楚适用范围：主动推理不是某个开箱即用的一键包，而是一种智能体上下文管理策略，所以下面表格里的每一项都取决于你的具体实现。比如选什么模型、用什么检索组件、是否本地部署，都会影响最终参数。下表是这套方案的通用能力画像，实际落地时按自己的技术栈逐项确认。

从材料看，目前社区讨论比较多的是“智能体、模型、AI、token 的关系”“人工智能 skills 怎么安装到智能体上”以及“如何编写 AI 智能体”，这些话题其实都落在这张表里：模型负责推理，token 是推理的计费单位，skills 是给智能体安装的外部能力，而主动推理要解决的是“模型每一次推理到底该带多少上下文”。

2. 主动推理是什么：从“全量塞入”到“按需获取”

2.1 智能体、模型、AI、token 的关系

先把四个概念理顺。AI 智能体是一个能调用模型和工具来完成任务的程序，模型是智能体的推理引擎，token 是模型处理和计费的最小文本单位。上下文越长，单次推理的 token 数越大，意味着更高的费用、更长的首字延迟，并且当无关内容太多时，模型的注意力会被稀释，反而容易答错。热词里反复出现“token 关系”问题，本质就是上下文预算问题：给少了不够用，给多了用不起。

主动推理的核心逻辑也在这里。它把“上下文预算”从开发者的静态配置，变成智能体的动态决策。智能体在推理过程中随时可以回答“当前还缺什么信息”，缺就去取，取完继续推理。这样 token 花在真正有用的内容上，而不是花在搬运一整份文档上。

2.2 被动上下文与主动推理的区别

这里要强调，主动推理的“按需”不是一次性把检索结果拼进 prompt，而是一个持续动作：先判断缺失信息，再调用检索或工具，拿到结果后继续判断，直到认为信息足够，再输出最终答案。整个过程类似“思考—取数—再思考”，而不是传统的“先取数—再回答”。

2.3 它与普通 RAG 的区别

很多人会把主动推理和 RAG 混在一起，两者有关联但不是一回事。常规 RAG 流程是在用户请求进入模型之前，由外部管道统一完成检索，然后把检索结果固定拼进 prompt，模型只负责基于这些内容做一次回答。这个流程适合“问知识库”这类一次性问答。

主动推理则把检索动作内化到智能体的推理循环中。它可以多次检索、按需切换数据源、在中间结果不充分时主动追加查询，甚至判断“这个问题根本不需要外部上下文”，然后跳过检索直接回答。RAG 可以看成主动推理的一个工具，但主动推理的决策范围更大，也更适合多步骤、多工具、长任务的场景。

3. 适用场景与使用边界

3.1 适合哪些场景

第一类是长文档问答。一份几百页的 PDF，如果全量进上下文，token 成本非常高，而且模型容易在无关章节里“迷失”；主动推理可以先看目录或摘要，确定问题指向哪个章节，再只取对应片段。

第二类是多来源信息汇总。比如比较两份合同、核对多个接口文档的差异，智能体需要从不同来源分别取数，再合并判断。这类任务天然适合按需获取，因为每一步需要的数据源都不同。

第三类是多工具串联。智能体需要先查订单系统，再调用库存接口，最后用计算工具得出结果。主动推理决定了每一步要访问哪个工具、把哪个工具的输出放进下一步上下文。

第四类是客服辅助和内部知识库搜索。用户问题往往涉及历史工单、产品文档、售后政策等多个数据源，主动推理能在较短的上下文里覆盖最相关的数据，降低每次客服会话的 token 成本。

3.2 不适合哪些场景

主动推理不适合单轮极低延迟的简单问答。比如用户问“现在几点”“1 加 1 等于几”，如果每个请求都先让模型做一轮“缺什么上下文”的决策，相当于给简单任务增加了额外的模型调用开销，延迟和成本反而更高。

也不适合必须保证上下文完整性的场景。合规审计、医疗诊断辅助、法律原文核对这类任务，不能只依赖检索结果，否则漏掉关键条款的后果很严重。这类场景应该保证原文完整可追溯，主动推理只能作为辅助定位手段，不能作为唯一信息源。

还有一点要特别注意：主动推理会频繁读取知识库、数据库、用户历史记录，这些数据可能包含隐私信息和企业敏感资料。在接入任何涉及人脸、声音、版权素材或个人数据的场景时，必须先确认授权范围，对上下文日志做脱敏处理，并严格控制服务访问权限。这也属于使用边界的一部分，不能等上线后再补。

4. 上下文管理常见方案对比

在做主动推理之前，先看看现在常见方案的优缺点，这样能理解为什么要多一层决策。

从这张表能看出，主动推理不是要替代其他方案，而是做在它们之上的一层决策。它可以决定“这次用滑动窗口就好”“这个问题需要做摘要”“这轮必须去向量库检索”。把决策层独立出来，是这套思路的关键。

5. 架构设计：主动推理在智能体中的落地方式

5.1 核心组件拆解

落地时，建议把智能体拆成六个组件，职责清晰，排查问题也方便。

任务解析器负责把用户输入拆成可执行目标，提取关键实体和约束条件。上下文决策器是主动推理的核心，它根据当前任务和已有上下文，输出“还需要什么”。上下文获取器负责访问数据源，可以是向量检索、文档读取、数据库查询或外部 API 调用。执行器负责调用模型完成生成或工具操作。验证器负责检查当前信息是否足够、结果是否正确。上下文缓存负责缓存已经取过的内容，避免同一轮任务里重复检索。

5.2 运行闭环

整个闭环可以描述为：感知当前状态，推理缺失信息，决定获取来源，通过工具或检索取数，执行生成或操作，验证结果，信息不足则回到推理环节继续补。

这个循环必须有终止条件。通常设置最大迭代次数，比如 3 到 5 轮，超过后强制输出当前结果或返回“信息不足”。否则，当检索结果总是不能满足决策器要求时，智能体会陷入无限检索，token 消耗比全量塞入还高，这是实践中最容易踩的坑。

5.3 触发条件

不是每个任务都需要完整走一轮主动推理。下面这些情况适合触发按需获取：

• 用户问题中包含未知实体或专有名词，模型无法仅凭已有上下文回答。
• 任务需要跨多轮引用历史结论，当前上下文里没有记录。
• 生成结果置信度过低，验证器认为需要补充证据。
• 用户明确要求“查一下”“看看文档里怎么说”。

反过来，简单寒暄、单轮常识问答、当前上下文已经足够时，应该直接回答，不做多余检索。

6. 实现思路与代码示例

下面给出的是通用模板，不是某个项目的一比一实现。你需要按实际使用的模型 SDK、检索服务和工具框架做替换。

6.1 智能体主循环骨架

# 通用模板：请按实际项目替换模型调用与检索实现 from dataclasses import dataclass, field @dataclass class AgentState: task: str current_context: list[str] = field(default_factory=list) missing: list[str] = field(default_factory=list) iterations: int = 0 class ActiveInferenceAgent: def __init__(self, llm, retriever, tools, cache=None, max_iterations=5): self.llm = llm self.retriever = retriever self.tools = tools self.cache = cache self.max_iterations = max_iterations def run(self, task: str) -> str: state = AgentState(task=task) while state.iterations < self.max_iterations: decision = self.llm.decide_context( task=state.task, current_context=state.current_context, ) if not decision.needs_more: return self.llm.answer( task=state.task, context=state.current_context, ) fetched = self.fetch(decision.what_to_fetch) state.current_context.extend(fetched) state.missing = decision.what_to_fetch state.iterations += 1 return self.llm.answer(task=state.task, context=state.current_context) def fetch(self, fetch_requests: list[dict]) -> list[str]: # 按 fetch_requests 中的来源和条件去实际获取内容 results = [] for req in fetch_requests: source = req.get("source") if source == "vector_store": results.extend(self.retriever.search(req["query"], top_k=3)) elif source == "document": results.append(self.retriever.get_document_section(req["doc_id"], req["section"])) return results

这里decide_context是示意方法，实际实现通常用 function calling 让模型输出结构化决策结果，而不是单独训练一个决策模型。你可以在模型支持函数调用的情况下，把“要不要继续取数、取哪些数据”定义成一个工具调用，让模型在对话中自然触发。

6.2 上下文需求决策模板

模型决策环节的输出，建议用 JSON 结构化，方便后续解析和执行。下面是一个示意结构：

{ "task": "总结这份发布说明中的性能优化点", "current_context": ["文档目录", "开头 3 段"], "analysis": "缺少性能优化章节的详细内容", "needs_more": true, "what_to_fetch": [ { "source": "document", "doc_id": "release_notes.md", "section": "performance" } ] }

实际使用中，what_to_fetch可以是向量检索的 query，也可以是某个工具的参数。把决策结果标准化，后面无论是做日志、审计，还是做批量任务的失败重试，都会方便很多。

6.3 以 function calling 接入模型

主动推理对模型的核心要求是能稳定输出结构化工具调用。以函数调用方式接入时，要把“按需获取上下文”本身定义成一组工具。下面是工具注册的示意代码：

# 以最常见的 function calling 风格为例 tools = [ { "type": "function", "function": { "name": "retrieve_context", "description": "根据任务需求从知识库获取上下文片段", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "要检索的关键内容"}, "source": {"type": "string", "enum": ["docs", "history", "database"]} }, "required": ["query"] } } }, { "type": "function", "function": { "name": "Answer", "description": "当前上下文已足够，基于已有信息生成最终回答", "parameters": { "type": "object", "properties": { "answer": {"type": "string", "description": "最终回答内容"} }, "required": ["answer"] } } } ] def call_llm_with_tools(messages, tools): # 通用模板：按实际模型 SDK 调整请求格式 response = llm.chat(messages=messages, tools=tools) return response

当模型返回retrieve_context调用时，智能体去执行检索，把结果追加到消息里，再次调用模型，直到模型返回Answer调用。

6.4 Skills 的安装与注册

社区里经常有人问“人工智能 skills 怎么安装到 AI 智能体上”。从实现来看，一个 skill 本质上就是一个打包好的能力单元，至少包含四部分：名称、描述、输入参数约束、执行函数。安装到智能体，就是把这段能力注册进它的工具列表。

SKILLS = {} def register_skill(name, description, parameters_schema, handler): SKILLS[name] = { "description": description, "parameters": parameters_schema, "handler": handler, } register_skill( name="retrieve_release_notes", description="按关键字检索发布说明文档，适合回答版本变更、性能优化相关问题", parameters_schema={ "type": "object", "properties": {"query": {"type": "string"}} }, handler=lambda query: retrieve_from_docs(query), )

skill 的描述质量直接影响主动推理的效果。描述写得太笼统，模型不知道该在什么时候调用；描述写得太窄，该调用时不调用。建议在描述里写明触发条件和典型场景，比如“适合回答版本变更、性能优化相关问题”。这一步是安装 skills 时最容易忽略的细节。

7. 功能测试与效果验证

主动推理的好坏不能只看“能不能答对”，还要看“为了答对花了多少 token、做了几次检索”。建议至少跑以下五类用例。

7.1 测试用例设计

7.2 关键评估指标

建议每次任务都记录四个指标：单任务 token 消耗、检索调用次数、迭代轮数、总耗时。再结合答案准确率一起看。

不要只看准确率。一个智能体如果每次任务都做 10 次检索，准确率可能很高，但 token 成本已经失去控制。主动推理的优化目标是“在可接受的准确率下，让 token 和检索次数最小化”。所以记录指标时，要把决策轮次和检索次数一起记录下来。

7.3 判断成功与失败

判断一次主动推理是否成功，可以看三个信号：

第一，该检索时是否检索了。缺失信息的任务如果没有触发检索，说明决策器没有正常工作，优先检查工具描述和模型调用格式。第二，不该检索时是否跳过。简单问题如果也做了大量检索，说明触发条件设置过宽，需要在决策提示词里加入“当前上下文足够时直接回答”的约束。第三，每轮决策是否收敛。如果迭代轮数总是逼近上限，说明获取到的内容重复或不足，需要改进检索质量和去重逻辑。

8. 接口 API 与批量任务

主动推理适合用于生产化智能体服务。把智能体封装成 HTTP 接口后，前端、工作流、其他后端服务都可以直接调用。

8.1 将智能体封装为 HTTP 服务

下面用 FastAPI 风格给一个通用模板。实际项目的鉴权、限流、日志都需要按企业标准补齐。

from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class TaskRequest(BaseModel): task: str session_id: str = "" @app.post("/agent/run") def run_agent(req: TaskRequest): # 实际实现需要注入已初始化的 agent 实例 result = agent.run(req.task) return {"session_id": req.session_id, "result": result}

8.2 curl 调用示例

curl -X POST http://127.0.0.1:8000/agent/run \ -H "Content-Type: application/json" \ -d '{"task": "总结 release notes 中的性能优化点", "session_id": "test-001"}'

如果单任务时间较长，不建议让客户端一直等待同步响应。更稳妥的做法是提交任务后返回 task_id，由客户端轮询任务状态，或者用 WebSocket 推送结果。主动推理有多次决策循环，单任务耗时通常比普通问答长，异步化能明显提升体验。

8.3 批量任务队列设计

批量任务的关键是：每个任务独立记录状态、异常要捕获、失败要重试。下面是一个简化模板。

import json import time def run_batch(input_file, output_file, max_retries=3): tasks = json.load(open(input_file, encoding="utf-8")) results = [] for task in tasks: ok = False for retry in range(max_retries): try: result = agent.run(task["question"]) results.append({"id": task["id"], "status": "ok", "result": result}) ok = True break except Exception as exc: if retry == max_retries - 1: results.append({"id": task["id"], "status": "failed", "error": str(exc)}) else: time.sleep(2 ** retry) # 指数退避 with open(output_file, "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2)

批量任务最好加上并发限制。主动推理的检索和模型调用都会产生资源消耗，无限制并发容易打满数据库连接或 LLM 服务的速率配额，导致整批任务大面积失败。

8.4 接口安全与合规

接口服务必须在网关层做身份认证和访问控制，只允许授权方调用。主动推理会读取知识库和用户数据，接口日志里可能包含敏感内容，所以日志要脱敏，检索数据源要按最小权限原则配置。这是接口化部署里不能省略的一步。

9. 资源占用与性能观察

主动推理的性能观察重点不是 GPU 显存，而是 token 和延迟。当然，如果你自己部署了本地大模型，显存占用同样要监控，但这属于模型底座层面的问题，与主动推理策略本身关系不大。

9.1 观察哪些指标

每轮任务建议输出一条结构化日志，包含任务 ID、迭代次数、决策调用 token 数、检索次数、总 token 数、总耗时。格式可以参考下面这样：

{ "task_id": "t-001", "iterations": 3, "decision_tokens": 860, "fetch_calls": 2, "total_tokens": 4210, "latency_ms": 4820 }

有了这类日志，你能很快判断问题出在哪：迭代次数高说明决策不收敛，fetch_calls 高说明检索太碎，total_tokens 高说明上下文拼接有浪费。

9.2 影响性能的关键因素

第一个因素是决策轮次。每多一轮决策，就要多一次模型调用，延迟和 token 都会增加。控制最大迭代次数是最直接的优化手段。

第二个因素是检索结果长度。每次检索可能返回多段内容，如果每段都完整塞进上下文，token 会快速增长。建议在获取器里对结果做长度截断或摘要，只保留最关键的部分。

第三个因素是缓存命中率。同一会话内重复查询相同内容时，如果缓存生效，可以省掉一次检索和一次模型决策。主动推理的多轮特性决定了同一个上下文可能被多次引用，缓存收益通常比普通 RAG 更明显。

9.3 如何降低开销

给检索结果加长度上限，超出部分截断或摘要；对决策结果做去重，避免同一轮迭代重复获取相同内容；把最大迭代次数从 5 降到 3，观察准确率是否有明显下降；引入 reranker 提升检索精度，减少无效 fetch 次数。这些手段要一个个试，不要一次性全上，否则很难判断哪个改动真正有效。

10. 常见问题与排查方法

11. 最佳实践与使用建议

第一次做主动推理，不要一上来就接十几个工具。建议先用一个最小闭环验证：一个检索工具、一个回答工具、最多三轮迭代。跑通后再逐渐加数据源和复杂逻辑。

上下文数据、输入素材、输出结果、日志要分目录管理。主动推理涉及多次检索和多轮决策，日志是最重要的排障依据。建议每条任务至少保留完整决策链，包括每次决策的 JSON、每次检索的 query 和返回片段长度。

给检索和工具调用都设置超时。主动推理的循环里，任何一次外部调用挂起，都会拖垮整个任务。超时参数要单独设置，不能依赖模型调用的默认超时。

发布前一定要做效果复核。主动推理模式下的答案可能来自检索片段拼接，模型可能会把不同来源的信息混合，生成看似合理但实际错误的结论。尤其涉及合同、数据报表、医疗信息时，必须人工检查关键结论和来源。

版权、隐私和授权问题不能忽略。知识库里的文档、用户历史记录、第三方接口数据，在使用前要确认是否有合法授权。涉及人脸、声音、版权素材时，必须明确授权边界。对外提供服务时，接口要加认证和限流，日志要脱敏，这是底线要求。

12. 总结与下一步

这套思路最值得尝试的点，是它改变了对上下文的管理方式：从“开发者替模型决定带什么”变成“模型自己决定缺什么”。第一次接入时，建议优先验证一个核心问题：决策器能不能准确识别缺失信息并触发检索。这一步跑通了，后面的多工具、批量任务才有意义。

最容易踩的坑是决策循环失控。要么模型始终不触发检索，导致关键信息缺失；要么模型反复检索，token 成本比全量塞入还高。解决方法是把决策输出标准化，记录每轮迭代的 token 和检索次数，用数据判断是否收敛。

如果你已经在做智能体应用，可以从一个只有两三个工具的小项目开始，先建立基线数据：每轮迭代次数、单任务 token 消耗、检索命中率。跑通之后，再逐步接入更多 skills、扩展成异步接口服务、加上批量任务队列。这样每一步改动都能用数据验证，不会越改越黑盒。