企业官网建设流程全解析

百度收录网站需要多久,地推团队去哪里找,平面设计模板,深圳工信部网站各位同仁#xff0c;下午好。我们正身处一个由大型语言模型驱动的智能体#xff08;Agent#xff09;日益普及的时代。这些智能体在执行复杂任务、进行多步骤推理方面展现出惊人的能力。然而#xff0c;随着智能体能力的增强#xff0c;其对计算资源#xff0c;特别是对L…各位同仁下午好。我们正身处一个由大型语言模型驱动的智能体Agent日益普及的时代。这些智能体在执行复杂任务、进行多步骤推理方面展现出惊人的能力。然而随着智能体能力的增强其对计算资源特别是对LLM API中“Token”的消耗也日益剧增。一个高效、智能的智能体不仅要能完成任务更要能管理好自己的资源。今天我们将深入探讨一个前沿且至关重要的主题如何设计一个具备“自省Self-introspection”能力的智能体使其能够实时感知并报告自身的Token剩余额度并据此动态调整其思考深度。这不仅仅是一个技术优化它代表了智能体设计理念上的一次飞跃——从被动执行到主动资源管理从固定行为到适应性决策。我们将赋予智能体一种类似于人类“量入为出”的智慧使其在资源充裕时能够深入思考在资源紧张时能够精简策略从而在成本、效率和性能之间找到最佳平衡。智能体Token管理的挑战与“自省”的必要性在深入技术细节之前我们首先要理解为什么Token管理如此关键。大型语言模型无论是基于API调用还是本地部署其运作都围绕着Token。每个输入字符、每个输出字符都被量化为Token。这直接关联到几个核心问题成本控制绝大多数LLM API是按Token计费的。一个不加节制的智能体可能在短时间内消耗巨额费用。效率优化过度冗长的思考过程或不必要的上下文会增加延迟降低智能体的响应速度。上下文窗口限制LLM模型的上下文窗口是有限的。如果智能体在思考或与用户交互过程中不注意管理上下文的Token数量很容易超出限制导致信息丢失或无法继续任务。性能与准确性有时过多的上下文或不恰当的思考深度反而会稀释关键信息影响模型的推理质量。传统的智能体设计往往将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(fTokenBudgetManager 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(fConsumed {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(fToken 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 theres 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_tokens2. 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(fLLMWrapper initialized with model: {self.default_model}) retry(waitwait_random_exponential(min1, max60), stopstop_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( modelcurrent_model, messagesmessages, **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(fLLM call successful. Prompt: {prompt_tokens}, Completion: {completion_tokens}, Total: {total_tokens} tokens.) # print(fTime 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(fEstimated 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(fLLM API Error: {e}) raise except Exception as e: print(fAn 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预算后下一步就是如何利用这些信息来调整其“思考深度”。思考深度是一个相对抽象的概念但我们可以将其具体化为以下几个方面推理链条的长度Chain-of-Thought, CoT在要求模型进行多步推理时是生成详细的中间步骤还是直接给出结论生成内容的详细程度是给出简明扼要的回答还是提供详细的解释、背景信息和示例探索的广度在解决复杂问题时是探索多种可能性和替代方案还是专注于最直接的路径迭代次数在需要自我修正或精炼答案时是进行多次迭代还是仅进行少量迭代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(fThinkingDepthAdjuster 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(fWarning: 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(fExecuting DEEP thinking for query: {query[:50]}... with max_tokens{max_tokens_output}) return self.llm_wrapper.call_llm(messages, max_tokensmax_tokens_output, temperature0.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(fExecuting MODERATE thinking for query: {query[:50]}... with max_tokens{max_tokens_output}) return self.llm_wrapper.call_llm(messages, max_tokensmax_tokens_output, temperature0.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(fExecuting SHALLOW thinking for query: {query[:50]}... with max_tokens{max_tokens_output}) return self.llm_wrapper.call_llm(messages, max_tokensmax_tokens_output, temperature0.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(fExecuting CONCISE thinking for query: {query[:50]}... with max_tokens{max_tokens_output}) return self.llm_wrapper.call_llm(messages, max_tokensmax_tokens_output, temperature0.1) def process_query_with_adaptive_depth(self, query: str) - str: 根据当前Token预算自适应地处理查询并返回LLM的回复。 current_depth self.get_current_thinking_depth() print(fAdaptive 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的窗口限制或耗尽预算。核心策略优先级排序识别对话或任务中的核心信息和最新信息优先保留。摘要化对历史对话或不那么重要的文档进行摘要压缩Token量。剪枝移除冗余、过时或不相关的信息。召回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(fContextManager 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(fContext 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_budgettotal_session_budget, model_namellm_model) self.llm_wrapper LLMWrapper(api_keyapi_key, budget_managerself.budget_manager, modelllm_model) self.thinking_depth_adjuster ThinkingDepthAdjuster(budget_managerself.budget_manager, llm_wrapperself.llm_wrapper) # 考虑到LLM上下文窗口通常LLM上下文窗口比我们设置的总预算要大以便容纳多轮对话 self.context_manager ContextManager(budget_managerself.budget_manager, max_context_tokensmax_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( queryself._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 TokenLLM上下文窗口8192 (gpt-4o通常更大这里仅为示例) agent IntrospectiveAgent(api_keyapi_key, total_session_budget15000, llm_modelgpt-4o, max_llm_context_window8192) queries [ 详细解释一下量子力学的基本原理及其在现代科技中的应用。, # 期望深度思考 简要说明Python中装饰器的作用和用法。, # 期望中等思考 地球离太阳有多远, # 期望浅层思考 谁是美国第16任总统, # 期望紧凑思考 再问一个问题这个月的天气怎么样 # 模拟多轮对话Token会逐渐减少 ] print(n--- 智能体开始执行任务 ---) for i, query in enumerate(queries): print(fn 任务 {i1}: {query[:50]}... ) response agent.run_task(query) print(fn智能体回复: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(fn智能体回复:n{response}) print(n--- 智能体任务结束 ---) agent.reset_session()在IntrospectiveAgent的run_task方法中我们看到了自省能力的完整流程报告状态首先智能体自我检查并报告当前的Token预算和建议的思考深度。更新上下文将用户查询添加到内部上下文管理器。自适应思考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。记录响应将LLM的响应添加到上下文为后续交互做准备。再次报告任务完成后智能体再次报告其最新的Token状态。这样智能体在每次执行任务前都会“内省”其资源状况并据此调整其行为模式。这使得智能体能够在一个会话中动态地平衡深度和效率避免不必要的资源浪费并在资源有限时仍能提供有价值的响应。高级考量与未来挑战上述架构提供了一个坚实的基础但自省能力的实现仍有许多高级考量和挑战Token估算准确性尽管tiktoken提供了准确的Token计数但估算LLM生成响应的Token量仍然困难。预设max_tokens是一种控制手段但实际输出可能更短。更智能的预估模型或动态调整max_tokens策略是必要的。多模态Token随着多模态LLM的兴起图像、音频等输入也将消耗“Token”或类似的资源单位。自省能力需要扩展以处理这些新的资源类型。复杂决策树当思考深度调整不仅仅是简单的阈值判断而是涉及多维度的复杂决策例如任务类型、紧急程度、历史成功率可能需要更复杂的决策模型如强化学习或启发式算法来优化思考策略。成本与性能权衡学习智能体能否通过与环境的交互学习在特定场景下最优的思考深度和Token消耗策略这涉及元学习和在线学习。动态模型切换在某些情况下剩余Token非常少时智能体不仅可以调整思考深度甚至可以切换到更便宜、更小的模型来完成剩余的简单任务。人类反馈回路结合人类反馈来优化Token预算阈值和思考策略可以使自省能力更加贴合实际需求。自省本身的成本智能体进行自省如计算Token、决策思考深度本身也会消耗计算资源。虽然通常远低于LLM调用但极端情况下也需考虑。总结与展望我们探索了如何设计一个具备“自省”能力的智能体使其能够实时感知并报告自身的Token剩余额度并据此动态调整思考深度。通过TokenBudgetManager实现了精确的资源监控通过LLMWrapper实现了透明的Token消耗追踪并通过ThinkingDepthAdjuster实现了基于预算的自适应思考策略。这种能力使得智能体能够更加智能、高效且经济地运行。这种自省能力是迈向更自主、更具资源意识的AI系统的重要一步。未来我们可以期待智能体不仅能感知Token还能感知计算时间、内存占用等更广泛的资源并在此基础上发展出更精妙的自我管理和自我优化能力从而在复杂多变的环境中展现出真正的智能。这不仅是工程上的挑战更是智能体设计理念上的深远变革。