深度挑战：设计一个具备‘自省（Self-introspection）’能力的 Agent，它能实时报告自己的 Token 剩余额度并调整思考深度

各位同仁，下午好。

我们正身处一个由大型语言模型驱动的智能体（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预算管理器	追踪、报告并管理智能体的Token预算。	API响应解析、tiktoken等分词器、内部计数器
LLM调用封装器	拦截所有对LLM的调用，执行Token计数与预算扣除，并处理API响应。	Python装饰器、代理模式、异步IO处理
上下文管理器	根据Token预算和任务需求，动态调整LLM输入的上下文内容。	摘要、剪枝、信息优先级排序、向量数据库检索
思考深度调整器	根据剩余Token预算，决定智能体下一步的推理策略和输出详细程度。	策略模式、参数配置（max_tokens、CoT步骤数）、提示词工程
智能体核心控制器	协调上述模块，根据任务目标、环境反馈和自省信息，驱动智能体行为。	状态机、规划器、决策树

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

模块实现一：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，还能感知计算时间、内存占用等更广泛的资源，并在此基础上发展出更精妙的自我管理和自我优化能力，从而在复杂多变的环境中展现出真正的智能。这不仅是工程上的挑战，更是智能体设计理念上的深远变革。