福建省网站建设_网站建设公司_UI设计师_seo优化

各位同仁，下午好。

我们正身处一个由大型语言模型驱动的智能体（Agent）日益普及的时代。这些智能体在执行复杂任务、进行多步骤推理方面展现出惊人的能力。然而，随着智能体能力的增强，其对计算资源，特别是对LLM API中“Token”的消耗也日益剧增。一个高效、智能的智能体，不仅要能完成任务，更要能管理好自己的资源。今天，我们将深入探讨一个前沿且至关重要的主题：如何设计一个具备“自省（Self-introspection）”能力的智能体，使其能够实时感知并报告自身的Token剩余额度，并据此动态调整其思考深度。

这不仅仅是一个技术优化，它代表了智能体设计理念上的一次飞跃——从被动执行到主动资源管理，从固定行为到适应性决策。我们将赋予智能体一种类似于人类“量入为出”的智慧，使其在资源充裕时能够深入思考，在资源紧张时能够精简策略，从而在成本、效率和性能之间找到最佳平衡。

智能体Token管理的挑战与“自省”的必要性

在深入技术细节之前，我们首先要理解为什么Token管理如此关键。大型语言模型，无论是基于API调用还是本地部署，其运作都围绕着Token。每个输入字符、每个输出字符，都被量化为Token。这直接关联到几个核心问题：

1. 成本控制：绝大多数LLM API是按Token计费的。一个不加节制的智能体可能在短时间内消耗巨额费用。
2. 效率优化：过度冗长的思考过程或不必要的上下文，会增加延迟，降低智能体的响应速度。
3. 上下文窗口限制：LLM模型的上下文窗口是有限的。如果智能体在思考或与用户交互过程中，不注意管理上下文的Token数量，很容易超出限制，导致信息丢失或无法继续任务。
4. 性能与准确性：有时，过多的上下文或不恰当的思考深度反而会稀释关键信息，影响模型的推理质量。

传统的智能体设计往往将Token管理视为外部问题，通过硬编码的限制或后处理来解决。然而，一个真正智能的智能体应该像人类一样，能够“感知”自身的思考成本，并根据当前的“预算”来决定如何“思考”。这种内部感知和适应性调整的能力，正是我们所说的“自省”。

一个具备自省能力的智能体，将拥有以下核心特征：

• 实时监控：能够精确追踪每一次LLM调用所消耗的输入和输出Token，并维护一个全局或任务级的Token预算。
• 状态报告：能够对外（或对自身更高层级的决策模块）报告当前可用的Token额度。
• 决策调整：基于剩余Token额度，动态调整其后续的思考策略、生成内容的详细程度、推理步骤的复杂性等。

架构设计：构建自省能力的基石

为了实现这种自省能力，我们需要对智能体的架构进行精心设计。核心思想是将Token管理和思考深度调整能力融入到智能体的核心决策流程中。以下是关键模块的概述：

接下来，我们将逐一深入这些模块的实现细节。

模块实现一：Token预算管理器与LLM调用封装器

这是自省能力的基础。我们需要一个可靠的机制来计算Token并管理预算。

1. Token预算管理器（TokenBudgetManager）

这个模块负责维护当前的Token预算，记录已消耗的Token，并提供查询剩余额度的方法。

import tiktoken from typing import Dict, Any, Optional class TokenBudgetManager: """ 管理智能体的Token预算。 """ def __init__(self, total_budget: int, model_name: str = "gpt-4o"): """ 初始化Token预算管理器。 :param total_budget: 智能体可用的总Token预算。 :param model_name: 用于计算Token的LLM模型名称，影响分词器选择。 """ if total_budget <= 0: raise ValueError("总Token预算必须是正数。") self._total_budget = total_budget self._consumed_tokens = 0 self._encoding = tiktoken.encoding_for_model(model_name) print(f"TokenBudgetManager initialized with total budget: {total_budget} tokens for model: {model_name}") def _count_tokens(self, text: str) -> int: """ 使用tiktoken库计算给定文本的Token数量。 :param text: 要计算Token的文本。 :return: Token数量。 """ return len(self._encoding.encode(text)) def consume_tokens(self, tokens_used: int): """ 从预算中扣除已使用的Token。 :param tokens_used: 本次操作消耗的Token数量。 """ if tokens_used < 0: raise ValueError("消耗的Token数量不能为负数。") self._consumed_tokens += tokens_used # print(f"Consumed {tokens_used} tokens. Current consumed: {self._consumed_tokens}") def get_remaining_budget(self) -> int: """ 获取当前剩余的Token预算。 :return: 剩余Token数量。 """ return self._total_budget - self._consumed_tokens def get_consumed_tokens(self) -> int: """ 获取已消耗的Token数量。 :return: 已消耗Token数量。 """ return self._consumed_tokens def get_total_budget(self) -> int: """ 获取总的Token预算。 :return: 总Token数量。 """ return self._total_budget def check_budget_exceeded(self) -> bool: """ 检查预算是否已超出。 :return: 如果预算已超出返回True，否则返回False。 """ return self.get_remaining_budget() <= 0 def reset_budget(self, new_total_budget: Optional[int] = None): """ 重置已消耗的Token数量，或设置新的总预算并重置。 :param new_total_budget: 可选，如果提供，则更新总预算。 """ self._consumed_tokens = 0 if new_total_budget is not None: if new_total_budget <= 0: raise ValueError("新的总Token预算必须是正数。") self._total_budget = new_total_budget print(f"Token budget reset. Current total budget: {self._total_budget}") def estimate_prompt_tokens(self, messages: list[Dict[str, str]]) -> int: """ 估算一个OpenAI风格的messages列表的Token数量。 参考：https://github.com/openai/openai-cookbook/blob/main/examples/How_to_count_tokens_with_tiktoken.ipynb """ tokens_per_message = 3 # every message follows <|start|>{role/name}n{content}<|end|>n tokens_per_name = 1 # if there's a name, the role is omitted num_tokens = 0 for message in messages: num_tokens += tokens_per_message for key, value in message.items(): num_tokens += self._count_tokens(value) if key == "name": num_tokens += tokens_per_name num_tokens += 3 # every reply is primed with <|start|>assistant<|message|> return num_tokens

2. LLM调用封装器（LLMWrapper）

这个封装器是智能体与LLM API交互的桥梁，它会在每次调用前后自动进行Token计数和预算更新。我们通常会使用openai这样的库。

import openai import time from tenacity import retry, wait_random_exponential, stop_after_attempt from functools import wraps class LLMWrapper: """ 一个封装LLM调用的类，负责与TokenBudgetManager交互，并处理API响应。 """ def __init__(self, api_key: str, budget_manager: TokenBudgetManager, model: str = "gpt-4o"): """ 初始化LLMWrapper。 :param api_key: OpenAI API密钥。 :param budget_manager: TokenBudgetManager实例。 :param model: 默认使用的LLM模型名称。 """ openai.api_key = api_key self.budget_manager = budget_manager self.default_model = model print(f"LLMWrapper initialized with model: {self.default_model}") @retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6)) def call_llm(self, messages: list[Dict[str, str]], model: Optional[str] = None, **kwargs: Any) -> Dict[str, Any]: """ 调用LLM模型并自动处理Token计数。 :param messages: 传递给LLM的对话消息列表。 :param model: 本次调用使用的模型，如果未指定则使用默认模型。 :param kwargs: 其他传递给openai.ChatCompletion.create的参数。 :return: LLM的响应字典。 """ current_model = model if model else self.default_model # 1. 估算Prompt Token并检查预算 prompt_tokens_estimate = self.budget_manager.estimate_prompt_tokens(messages) if self.budget_manager.get_remaining_budget() < prompt_tokens_estimate: raise ValueError(f"调用LLM前预算不足。需要 {prompt_tokens_estimate} tokens，剩余 {self.budget_manager.get_remaining_budget()} tokens。") # 2. 调用LLM API start_time = time.time() try: response = openai.chat.completions.create( model=current_model, messages=messages, **kwargs ) end_time = time.time() # 3. 解析API响应中的Token使用量并更新预算 if response.usage: prompt_tokens = response.usage.prompt_tokens completion_tokens = response.usage.completion_tokens total_tokens = response.usage.total_tokens # 理论上可以直接用total_tokens，但为明确起见，我们分别记录 self.budget_manager.consume_tokens(total_tokens) # print(f"LLM call successful. Prompt: {prompt_tokens}, Completion: {completion_tokens}, Total: {total_tokens} tokens.") # print(f"Time taken: {end_time - start_time:.2f}s. Remaining budget: {self.budget_manager.get_remaining_budget()} tokens.") else: # 如果API未能返回usage信息，我们进行保守估算 print("Warning: LLM API did not return usage information. Estimating tokens.") # 估算输入Token estimated_prompt_tokens = self.budget_manager.estimate_prompt_tokens(messages) # 估算输出Token（假设平均每个token 4个字符） estimated_completion_tokens = self.budget_manager._count_tokens(response.choices[0].message.content) if response.choices else 0 estimated_total_tokens = estimated_prompt_tokens + estimated_completion_tokens self.budget_manager.consume_tokens(estimated_total_tokens) print(f"Estimated token usage: {estimated_total_tokens}. Remaining budget: {self.budget_manager.get_remaining_budget()} tokens.") return response.to_dict() # Convert Pydantic model to dict for consistency except openai.APIError as e: print(f"LLM API Error: {e}") raise except Exception as e: print(f"An unexpected error occurred during LLM call: {e}") raise def get_current_model(self) -> str: """获取当前封装器使用的默认模型名称。""" return self.default_model def get_remaining_budget_report(self) -> str: """生成一个关于剩余预算的报告字符串。""" return (f"当前Token预算报告：n" f" 总预算: {self.budget_manager.get_total_budget()} tokensn" f" 已消耗: {self.budget_manager.get_consumed_tokens()} tokensn" f" 剩余: {self.budget_manager.get_remaining_budget()} tokens")

通过LLMWrapper，所有的LLM交互都将自动通过TokenBudgetManager进行Token的计算和扣除，从而实现了智能体的实时Token报告能力。任何时候，我们都可以通过budget_manager.get_remaining_budget()来获取当前剩余的Token额度。

模块实现二：思考深度调整器

当智能体能够实时感知其Token预算后，下一步就是如何利用这些信息来调整其“思考深度”。思考深度是一个相对抽象的概念，但我们可以将其具体化为以下几个方面：

1. 推理链条的长度（Chain-of-Thought, CoT）：在要求模型进行多步推理时，是生成详细的中间步骤，还是直接给出结论？
2. 生成内容的详细程度：是给出简明扼要的回答，还是提供详细的解释、背景信息和示例？
3. 探索的广度：在解决复杂问题时，是探索多种可能性和替代方案，还是专注于最直接的路径？
4. 迭代次数：在需要自我修正或精炼答案时，是进行多次迭代，还是仅进行少量迭代？
5. LLM参数调整：例如，max_tokens参数直接控制输出长度，而temperature间接影响思考的“发散性”。

我们将通过一个ThinkingDepthAdjuster类来实现这一功能，该类将根据TokenBudgetManager提供的剩余预算，决定采取哪种思考策略。

定义思考策略

我们首先定义几种思考策略，它们将对应不同的提示词工程和LLM参数设置。

from enum import Enum class ThinkingDepth(Enum): """定义智能体的思考深度级别。""" DEEP = "DEEP" # 深度思考：详细的推理、多步CoT、丰富的内容 MODERATE = "MODERATE" # 中等思考：简明的推理、关键CoT步骤、主要内容 SHALLOW = "SHALLOW" # 浅层思考：直接的答案、无CoT或极简CoT、精炼内容 CONCISE = "CONCISE" # 紧凑思考：最直接的答案，可能牺牲部分准确性以节省Token class ThinkingDepthAdjuster: """ 根据Token预算动态调整智能体的思考深度。 """ def __init__(self, budget_manager: TokenBudgetManager, llm_wrapper: LLMWrapper): self.budget_manager = budget_manager self.llm_wrapper = llm_wrapper # 定义不同思考深度所需的Token阈值 (这是一个经验值，需要根据实际模型和任务调整) # 假设： # DEEP_THRESHOLD: 允许进行深度思考的最低剩余Token # MODERATE_THRESHOLD: 允许进行中等思考的最低剩余Token # SHALLOW_THRESHOLD: 允许进行浅层思考的最低剩余Token # CONCISE_THRESHOLD: 允许进行紧凑思考的最低剩余Token # 注意：这些阈值应该与实际的prompt长度和预期回复长度相匹配 # 这里我们使用一个相对值，具体应用时可能需要更精细的计算 self.thresholds: Dict[ThinkingDepth, int] = { ThinkingDepth.DEEP: int(budget_manager.get_total_budget() * 0.6), # 剩余60%以上 ThinkingDepth.MODERATE: int(budget_manager.get_total_budget() * 0.3), # 剩余30%以上 ThinkingDepth.SHALLOW: int(budget_manager.get_total_budget() * 0.1), # 剩余10%以上 ThinkingDepth.CONCISE: int(budget_manager.get_total_budget() * 0.02) # 剩余2%以上，为避免OOM的最低限度 } print(f"ThinkingDepthAdjuster initialized with thresholds: {self.thresholds}") def get_current_thinking_depth(self) -> ThinkingDepth: """ 根据当前剩余的Token预算决定当前的思考深度。 """ remaining_tokens = self.budget_manager.get_remaining_budget() if remaining_tokens >= self.thresholds[ThinkingDepth.DEEP]: return ThinkingDepth.DEEP elif remaining_tokens >= self.thresholds[ThinkingDepth.MODERATE]: return ThinkingDepth.MODERATE elif remaining_tokens >= self.thresholds[ThinkingDepth.SHALLOW]: return ThinkingDepth.SHALLOW elif remaining_tokens >= self.thresholds[ThinkingDepth.CONCISE]: return ThinkingDepth.CONCISE else: print(f"Warning: Token budget critically low ({remaining_tokens} tokens). Forcing CONCISE thinking or risking failure.") return ThinkingDepth.CONCISE # 已经非常低了，只能勉强给出答案 def _generate_deep_thought(self, query: str) -> Dict[str, Any]: """ 执行深度思考的Prompt策略。 特点：详细的Chain-of-Thought，探索多种可能性，生成详细解释。 """ messages = [ {"role": "system", "content": "你是一个高度智能的AI助手，请提供最详尽、最深入的分析和推理过程。展示你的完整思考链条，考虑所有相关因素，并提供全面的解释。"}, {"role": "user", "content": f"请详细分析并回答以下问题：{query}nn请逐步思考：n1. 明确问题。n2. 识别关键概念和背景信息。n3. 提出可能的解决方案或解释。n4. 对每个方案进行深入评估。n5. 总结并给出最终答案，附带详细的推理过程。"} ] # 深度思考可以允许更长的输出 max_tokens_output = min(1500, self.budget_manager.get_remaining_budget() - self.budget_manager.estimate_prompt_tokens(messages) - 50) max_tokens_output = max(100, max_tokens_output) # 确保至少有一定输出 print(f"Executing DEEP thinking for query: '{query[:50]}...' with max_tokens={max_tokens_output}") return self.llm_wrapper.call_llm(messages, max_tokens=max_tokens_output, temperature=0.7) def _generate_moderate_thought(self, query: str) -> Dict[str, Any]: """ 执行中等思考的Prompt策略。 特点：简明的CoT，聚焦核心问题，提供主要解释。 """ messages = [ {"role": "system", "content": "你是一个智能的AI助手，请提供清晰、有逻辑的分析和推理。展示关键的思考步骤，并给出明确的答案。"}, {"role": "user", "content": f"请分析并回答以下问题：{query}nn请逐步思考：n1. 理解问题核心。n2. 提出主要观点。n3. 给出支持论据。n4. 总结答案。"} ] max_tokens_output = min(800, self.budget_manager.get_remaining_budget() - self.budget_manager.estimate_prompt_tokens(messages) - 50) max_tokens_output = max(80, max_tokens_output) print(f"Executing MODERATE thinking for query: '{query[:50]}...' with max_tokens={max_tokens_output}") return self.llm_wrapper.call_llm(messages, max_tokens=max_tokens_output, temperature=0.5) def _generate_shallow_thought(self, query: str) -> Dict[str, Any]: """ 执行浅层思考的Prompt策略。 特点：直接给出答案，可能包含一两句解释，避免冗长推理。 """ messages = [ {"role": "system", "content": "你是一个高效的AI助手，请直接给出问题的答案，并附带简要的解释。"}, {"role": "user", "content": f"请回答以下问题：{query}"} ] max_tokens_output = min(300, self.budget_manager.get_remaining_budget() - self.budget_manager.estimate_prompt_tokens(messages) - 50) max_tokens_output = max(60, max_tokens_output) print(f"Executing SHALLOW thinking for query: '{query[:50]}...' with max_tokens={max_tokens_output}") return self.llm_wrapper.call_llm(messages, max_tokens=max_tokens_output, temperature=0.3) def _generate_concise_thought(self, query: str) -> Dict[str, Any]: """ 执行紧凑思考的Prompt策略。 特点：仅给出最核心的答案，不包含任何额外解释或推理。 """ messages = [ {"role": "system", "content": "你是一个极简的AI助手，请仅用最少的文字直接给出问题的答案，不要包含任何额外信息或解释。"}, {"role": "user", "content": f"请直接给出以下问题的答案：{query}"} ] max_tokens_output = min(100, self.budget_manager.get_remaining_budget() - self.budget_manager.estimate_prompt_tokens(messages) - 20) max_tokens_output = max(20, max_tokens_output) # 确保至少有20个Token输出，避免过短 print(f"Executing CONCISE thinking for query: '{query[:50]}...' with max_tokens={max_tokens_output}") return self.llm_wrapper.call_llm(messages, max_tokens=max_tokens_output, temperature=0.1) def process_query_with_adaptive_depth(self, query: str) -> str: """ 根据当前Token预算，自适应地处理查询并返回LLM的回复。 """ current_depth = self.get_current_thinking_depth() print(f"Adaptive thinking: Current depth set to {current_depth.name}.") try: if current_depth == ThinkingDepth.DEEP: response = self._generate_deep_thought(query) elif current_depth == ThinkingDepth.MODERATE: response = self._generate_moderate_thought(query) elif current_depth == ThinkingDepth.SHALLOW: response = self._generate_shallow_thought(query) else: # ThinkingDepth.CONCISE response = self._generate_concise_thought(query) return response['choices'][0]['message']['content'] except ValueError as e: # 当预算不足以生成任何深度时，会抛出此异常 return f"错误：Token预算极低，无法执行任何思考。{e}" except Exception as e: return f"处理查询时发生错误：{e}"

在ThinkingDepthAdjuster中，我们定义了不同思考深度对应的Token阈值。这些阈值是关键，它们决定了智能体在何种资源状况下采取何种策略。在实际应用中，这些阈值需要根据具体的任务、LLM模型以及预期的平均输入/输出Token量进行细致的校准。

_generate_deep_thought等方法内部封装了不同思考深度的Prompt策略和LLM参数（如max_tokens、temperature）。这使得智能体能够根据自省结果，动态地切换其与LLM交互的方式。

模块实现三：上下文管理器（简述）

虽然本文主要聚焦于Token报告和思考深度调整，但上下文管理器与Token预算紧密相关。当智能体进行多轮对话或处理大量信息时，上下文Token很容易超出LLM的窗口限制或耗尽预算。

核心策略：

1. 优先级排序：识别对话或任务中的核心信息和最新信息，优先保留。
2. 摘要化：对历史对话或不那么重要的文档进行摘要，压缩Token量。
3. 剪枝：移除冗余、过时或不相关的信息。
4. 召回（Retrieval）：将大部分历史信息存储在向量数据库中，在需要时仅召回相关片段，而非将所有历史信息都塞入Prompt。

一个简化的上下文管理器可以这样与TokenBudgetManager协作：

class ContextManager: """ 管理智能体的上下文，根据Token预算进行优化。 """ def __init__(self, budget_manager: TokenBudgetManager, max_context_tokens: int): self.budget_manager = budget_manager self.max_context_tokens = max_context_tokens # LLM的上下文窗口限制 self.current_context: list[Dict[str, str]] = [] print(f"ContextManager initialized with max context tokens: {max_context_tokens}") def add_message_to_context(self, message: Dict[str, str]): """ 向当前上下文添加一条消息。 """ self.current_context.append(message) self._prune_context_if_needed() def get_context_for_llm(self) -> list[Dict[str, str]]: """ 获取用于LLM调用的优化后的上下文。 """ # 在实际应用中，这里会进行更复杂的摘要、剪枝逻辑 # 为了演示，我们仅确保总上下文Token不超过max_context_tokens self._prune_context_if_needed() return self.current_context def _prune_context_if_needed(self): """ 如果当前上下文超出最大Token限制，则进行剪枝。 这里仅进行简单的头部剪枝，实际应用中应更智能。 """ current_tokens = self.budget_manager.estimate_prompt_tokens(self.current_context) while current_tokens > self.max_context_tokens and len(self.current_context) > 1: # 移除最老的消息，但保留系统消息（如果存在） if self.current_context[0]['role'] == 'system': self.current_context.pop(1) # 移除第一条非系统消息 else: self.current_context.pop(0) # 移除最老的消息 current_tokens = self.budget_manager.estimate_prompt_tokens(self.current_context) # print(f"Context pruned. Current context tokens: {current_tokens}/{self.max_context_tokens}") def clear_context(self): """ 清空所有上下文。 """ self.current_context = [] print("Context cleared.")

这个ContextManager在每次添加消息时都会尝试剪枝，确保传递给LLM的上下文不会过长。它与TokenBudgetManager紧密协作，通过estimate_prompt_tokens方法来计算当前上下文的Token量。

智能体核心控制器：整合与驱动

现在，我们将所有模块整合到一个IntrospectiveAgent类中，它将是智能体的大脑，协调各个部分协同工作。

import os class IntrospectiveAgent: """ 一个具备自省能力的智能体，能够实时报告Token使用并调整思考深度。 """ def __init__(self, api_key: str, total_session_budget: int, llm_model: str = "gpt-4o", max_llm_context_window: int = 8192): self.budget_manager = TokenBudgetManager(total_budget=total_session_budget, model_name=llm_model) self.llm_wrapper = LLMWrapper(api_key=api_key, budget_manager=self.budget_manager, model=llm_model) self.thinking_depth_adjuster = ThinkingDepthAdjuster(budget_manager=self.budget_manager, llm_wrapper=self.llm_wrapper) # 考虑到LLM上下文窗口，通常LLM上下文窗口比我们设置的总预算要大，以便容纳多轮对话 self.context_manager = ContextManager(budget_manager=self.budget_manager, max_context_tokens=max_llm_context_window) # 初始化系统消息作为上下文的一部分 self.context_manager.add_message_to_context({"role": "system", "content": "你是一个智能的、资源意识强的AI助手。"}) print("IntrospectiveAgent initialized.") def _report_status(self): """报告当前Token使用情况和思考深度。""" print("n--- Agent Self-Introspection Report ---") print(self.llm_wrapper.get_remaining_budget_report()) print(f"当前思考深度建议: {self.thinking_depth_adjuster.get_current_thinking_depth().name}") print("--------------------------------------n") def run_task(self, user_query: str) -> str: """ 执行一个任务，包括自省、调整思考深度和生成响应。 """ self._report_status() # 将用户查询添加到上下文 self.context_manager.add_message_to_context({"role": "user", "content": user_query}) # 获取当前上下文，并在内部进行剪枝 current_llm_context = self.context_manager.get_context_for_llm() # 根据预算调整思考深度，并生成响应 # 这里的thinking_depth_adjuster.process_query_with_adaptive_depth # 会在内部调用llm_wrapper.call_llm，从而自动更新budget_manager print("Agent is processing query with adaptive thinking...") response_content = self.thinking_depth_adjuster.process_query_with_adaptive_depth( query=self._prepare_adaptive_prompt(user_query, current_llm_context) ) # 将智能体的响应也添加到上下文，以便下一轮对话 self.context_manager.add_message_to_context({"role": "assistant", "content": response_content}) self._report_status() return response_content def _prepare_adaptive_prompt(self, user_query: str, current_context: list[Dict[str, str]]) -> str: """ 根据上下文和用户查询准备最终的Prompt字符串，供thinking_depth_adjuster使用。 注意：这里的thinking_depth_adjuster内部会重新构建messages列表， 所以这里的current_context只是作为信息传入，而不是直接传给LLM的messages参数。 """ # 我们可以将完整的上下文作为背景信息提供给下一个Prompt # 这是一个简化处理，更复杂的场景应该让thinking_depth_adjuster直接处理messages列表 context_str = "n".join([f"{msg['role']}: {msg['content']}" for msg in current_context if msg['role'] != 'system']) if context_str: return f"历史对话/背景信息:n{context_str}nn我的问题是: {user_query}" return user_query def reset_session(self, new_total_budget: Optional[int] = None): """ 重置智能体状态，包括Token预算和上下文。 """ self.budget_manager.reset_budget(new_total_budget) self.context_manager.clear_context() self.context_manager.add_message_to_context({"role": "system", "content": "你是一个智能的、资源意识强的AI助手。"}) print("Agent session reset.") # --- 示例运行 --- if __name__ == "__main__": # 请替换为你的OpenAI API Key # os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" api_key = os.getenv("OPENAI_API_KEY") if not api_key: print("错误：请设置 OPENAI_API_KEY 环境变量或直接在代码中提供。") exit() # 初始化智能体，总预算15000 Token，LLM上下文窗口8192 (gpt-4o通常更大，这里仅为示例) agent = IntrospectiveAgent(api_key=api_key, total_session_budget=15000, llm_model="gpt-4o", max_llm_context_window=8192) queries = [ "详细解释一下量子力学的基本原理及其在现代科技中的应用。", # 期望深度思考 "简要说明Python中装饰器的作用和用法。", # 期望中等思考 "地球离太阳有多远？", # 期望浅层思考 "谁是美国第16任总统？", # 期望紧凑思考 "再问一个问题，这个月的天气怎么样？" # 模拟多轮对话，Token会逐渐减少 ] print("n--- 智能体开始执行任务 ---") for i, query in enumerate(queries): print(f"n======== 任务 {i+1}: {query[:50]}... ========") response = agent.run_task(query) print(f"n智能体回复:n{response}") # 模拟中间有其他操作或等待，消耗一些预算 if i == 1: print("n模拟中间消耗了500 Token...") agent.budget_manager.consume_tokens(500) agent._report_status() if agent.budget_manager.check_budget_exceeded(): print("n!!! Token预算已用尽，智能体无法继续执行深度任务。!!!") break time.sleep(1) # 模拟思考间隔 # 尝试在预算非常低的情况下再进行一次查询 if not agent.budget_manager.check_budget_exceeded(): print("n======== 任务X: 预算不足，尝试最简回答 ========") # 强制消耗大量Token，模拟预算紧张 remaining = agent.budget_manager.get_remaining_budget() if remaining > 1000: # 确保至少留下一些 agent.budget_manager.consume_tokens(remaining - 500) response = agent.run_task("总结一下人工智能的未来发展趋势。") print(f"n智能体回复:n{response}") print("n--- 智能体任务结束 ---") agent.reset_session()

在IntrospectiveAgent的run_task方法中，我们看到了自省能力的完整流程：

1. 报告状态：首先，智能体自我检查并报告当前的Token预算和建议的思考深度。
2. 更新上下文：将用户查询添加到内部上下文管理器。
3. 自适应思考：thinking_depth_adjuster.process_query_with_adaptive_depth方法被调用。这个方法会：
   • 再次查询budget_manager获取最新的剩余Token。
   • 根据预设阈值决定当前最佳的ThinkingDepth。
   • 根据选定的深度，构建相应的Prompt（包含CoT指令、详细度要求等）和LLM参数（max_tokens,temperature）。
   • 通过llm_wrapper.call_llm调用LLM API。
   • llm_wrapper会在API返回后，自动从budget_manager中扣除本次调用消耗的Token。
4. 记录响应：将LLM的响应添加到上下文，为后续交互做准备。
5. 再次报告：任务完成后，智能体再次报告其最新的Token状态。

这样，智能体在每次执行任务前都会“内省”其资源状况，并据此调整其行为模式。这使得智能体能够在一个会话中动态地平衡深度和效率，避免不必要的资源浪费，并在资源有限时仍能提供有价值的响应。

高级考量与未来挑战

上述架构提供了一个坚实的基础，但自省能力的实现仍有许多高级考量和挑战：

1. Token估算准确性：尽管tiktoken提供了准确的Token计数，但估算LLM生成响应的Token量仍然困难。预设max_tokens是一种控制手段，但实际输出可能更短。更智能的预估模型或动态调整max_tokens策略是必要的。
2. 多模态Token：随着多模态LLM的兴起，图像、音频等输入也将消耗“Token”或类似的资源单位。自省能力需要扩展以处理这些新的资源类型。
3. 复杂决策树：当思考深度调整不仅仅是简单的阈值判断，而是涉及多维度的复杂决策（例如：任务类型、紧急程度、历史成功率），可能需要更复杂的决策模型，如强化学习或启发式算法来优化思考策略。
4. 成本与性能权衡学习：智能体能否通过与环境的交互，学习在特定场景下最优的思考深度和Token消耗策略？这涉及元学习和在线学习。
5. 动态模型切换：在某些情况下，剩余Token非常少时，智能体不仅可以调整思考深度，甚至可以切换到更便宜、更小的模型来完成剩余的简单任务。
6. 人类反馈回路：结合人类反馈来优化Token预算阈值和思考策略，可以使自省能力更加贴合实际需求。
7. 自省本身的成本：智能体进行自省（如计算Token、决策思考深度）本身也会消耗计算资源。虽然通常远低于LLM调用，但极端情况下也需考虑。

总结与展望

我们探索了如何设计一个具备“自省”能力的智能体，使其能够实时感知并报告自身的Token剩余额度，并据此动态调整思考深度。通过TokenBudgetManager实现了精确的资源监控，通过LLMWrapper实现了透明的Token消耗追踪，并通过ThinkingDepthAdjuster实现了基于预算的自适应思考策略。这种能力使得智能体能够更加智能、高效且经济地运行。

这种自省能力是迈向更自主、更具资源意识的AI系统的重要一步。未来，我们可以期待智能体不仅能感知Token，还能感知计算时间、内存占用等更广泛的资源，并在此基础上发展出更精妙的自我管理和自我优化能力，从而在复杂多变的环境中展现出真正的智能。这不仅是工程上的挑战，更是智能体设计理念上的深远变革。