AI Agent智能体技术：从问答到执行的范式革命

标签：AI Agent、大模型、智能体、LangChain、ReAct、Function Calling📖 前言2026年5月20日，谷歌I/O 2026大会在美国加州山景城开幕。谷歌CEO桑达尔·皮查伊（Sundar Pichai）在大会上宣布：“我们已正式进入’智能体Gemini时代’。”就在同一天，百度Create 2026大会上，百度创始人李彦宏提出AI时代的“度量衡”——DAA（Daily Active Agents，日活智能体数），标志着AI产业从“参数竞赛”正式转向“价值验证”阶段。从Google的Gemini Spark到百度的DuMate，从Anthropic的Claude Code到OpenAI的GPT-5.5，**AI Agent（智能体）**已从概念走向成熟商用，成为2026年最炙手可热的技术方向。本文将深入剖析AI Agent的技术架构、核心算法、工程实现，并提供完整的代码示例，帮助开发者快速掌握这一革命性技术。一、AI Agent的本质：感知-规划-行动-反思闭环1.1 什么是AI Agent？AI Agent（智能体）是一种能够自主理解目标、规划行动路径、调用外部工具、执行复杂任务的AI系统。与传统的问答式AI不同，Agent具备：自主决策能力：根据环境反馈动态调整执行策略工具调用能力：通过Function Calling/API与外部系统交互长期记忆能力：跨会话保持上下文和学习成果自我反思能力：评估执行效果并持续优化用一句话概括：AI Agent = 大脑（LLM）+ 记忆 + 工具 + 规划引擎1.2 从“Chat”到“Act”的范式转移传统ChatBot的交互模式：用户输入 → LLM生成回答 → 结束AI Agent的交互模式：用户输入 → 理解目标 → 分解任务 → 调用工具 → 观察结果 → 自我反思 → 完成任务这种ReAct（Reasoning + Acting）循环让AI从被动回答者转变为主动执行者。谷歌将这种转变称为“2026年的痛点从’幻觉’转向’懒惰’——用户不想看一大段总结，他们想要结果。”1.3 Agent的核心价值维度传统AIAI Agent交互方式问答任务执行工具使用无Function Calling/API上下文单次会话长期记忆错误处理返回错误自我反思重试价值交付信息可执行结果二、AI Agent技术架构深度解析2.1 整体架构概览一个完整的AI Agent系统包含四大核心层：┌─────────────────────────────────────────────────────────────┐ │ 交互层（Interaction Layer） │ │ 对话UI │ API接口 │ 多模态输入 │ Webhook │ 定时调度 │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ Agent核心层（Agent Core Layer） │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────────┐ │ │ │ 规划模块 │ → │ 推理模块 │ → │ LLM大脑 │ ← │ 工具选择模块 │ │ │ │Planning │ │Reasoning│ │ LLM │ │Tool Select │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────────┘ │ │ ┌─────────┐ ┌─────────┐ ┌─────────────────────────┐ │ │ │记忆模块 │ │执行模块 │ │ 自我反思模块 │ │ │ │ Memory │ │Execute │ │ Self-Reflection │ │ │ └─────────┘ └─────────┘ └─────────────────────────┘ │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 工具层（Tools Layer） │ │ 搜索 │ 计算器 │ 代码执行 │ API调用 │ 文件操作 │ 数据库 │ 邮件 │ │ MCP协议（工具标准） │ └─────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────────────────────────────────────────────────┐ │ 数据层（Data Knowledge Layer） │ │ 向量数据库 │ 知识图谱 │ 文档库 │ Redis缓存 │ RAG引擎 │ └─────────────────────────────────────────────────────────────┘2.2 感知模块（Perception Module）感知模块负责处理多模态输入，包括文本、语音、图像等，并将其转换为标准化的观测数据。# Python实现：多模态输入感知模块fromdataclassesimportdataclassfromtypingimportList,Optional,Dict,AnyfromenumimportEnumimportjsonclassInputType(Enum):TEXT="text"VOICE="voice"IMAGE="image"FILE="file"API_EVENT="api_event"SCHEDULED="scheduled"@dataclassclassObservation:"""标准化观测数据"""input_type:InputType content:strraw_data:Any metadata:Dict[str,Any]timestamp:floatsession_id:struser_id:Optional[str]=NoneclassPerceptionModule:""" 感知模块：多模态信息输入与处理 核心功能： 1. 输入标准化 2. 去噪与整理 3. 高质量预处理 """def__init__(self):self.input_handlers={InputType.TEXT:self._handle_text,InputType.VOICE:self._handle_voice,InputType.IMAGE:self._handle_image,InputType.FILE:self._handle_file,InputType.API_EVENT:self._handle_api_event,InputType.SCHEDULED:self._handle_scheduled,}defperceive(self,raw_input:Any,input_type:InputType,session_id:str,user_id:Optional[str]=None)-Observation:""" 主入口：处理各类输入并返回标准化观测 """# 1. 选择对应的处理器handler=self.input_handlers.get(input_type)ifnothandler:raiseValueError(f"Unsupported input type:{input_type}")# 2. 调用处理器content,metadata=handler(raw_input)# 3. 构建标准化观测observation=Observation(input_type=input_type,content=content,raw_data=raw_input,metadata=metadata,timestamp=time.time(),session_id=session_id,user_id=user_id)returnobservationdef_handle_text(self,raw_input:str)-tuple[str,Dict]:"""处理文本输入"""# 文本清洗cleaned=raw_input.strip()# 意图识别元数据metadata={"length":len(cleaned),"has_url":"http"incleaned,"has_code":"```"incleaned,"language":detect_language(cleaned)}returncleaned,metadatadef_handle_voice(self,raw_input:bytes)-tuple[str,Dict]:"""处理语音输入 - 需要ASR"""# 实际项目中调用语音识别服务# asr_result = asr_service.recognize(raw_input)asr_result="语音转文字的识别结果"# 模拟metadata={"duration":len(raw_input),# 模拟"sample_rate":16000,"confidence":0.95}returnasr_result,metadatadef_handle_image(self,raw_input:bytes)-tuple[str,Dict]:"""处理图像输入"""# OCR识别# text = ocr_service.extract(raw_input)extracted_text="图像中的文字内容"# 模拟metadata={"image_size":len(raw_input),"format":"png/jpeg","ocr_confidence":0.92}returnextracted_text,metadatadef_handle_file(self,raw_input:Dict)-tuple[str,Dict]:"""处理文件上传"""# 解析文件内容file_path=raw_input.get("path")file_type=raw_input.get("type")content=self._parse_file(file_path,file_type)metadata={"file_name":raw_input.get("name"),"file_size":raw_input.get("size"),"file_type":file_type}returncontent,metadatadef_handle_api_event(self,raw_input:Dict)-tuple[str,Dict]:"""处理API事件触发"""event_type=raw_input.get("event_type")event_data=raw_input.get("data",{})content=f"Event:{event_type}, Data:{json.dumps(event_data)}"metadata={"event_source":raw_input.get("source"),"event_id":raw_input.get("id")}returncontent,metadatadef_handle_scheduled(self,raw_input:Dict)-tuple[str,Dict]:"""处理定时任务触发"""task_name=raw_input.get("task_name")schedule_time=raw_input.get("schedule_time")content=f"Scheduled task:{task_name}"metadata={"schedule_type":raw_input.get("type"),"schedule_time":schedule_time}returncontent,metadata2.3 记忆模块（Memory Module）记忆是Agent从“一次性工具”进化为“持续学习助手”的关键。系统采用分层记忆架构：# Python实现：分层记忆系统fromtypingimportList,Optional,Dict,Anyimporttimeimporttiktokenfromdataclassesimportdataclass,field@dataclassclassMemoryItem:"""记忆单元"""content:strmemory_type:str# 'short_term' | 'long_term' | 'experience'timestamp:floatimportance:float=0.5# 0-1，越高越重要access_count:int=0embedding:Optional[List[float]]=NoneclassShortTermMemory:""" 短期记忆：工作记忆，维护当前对话上下文 使用滑动窗口机制，避免上下文溢出 """def__init__(self,max_tokens:int=4000):self.buffer:List[MemoryItem]=[]self.max_tokens=max_tokens self.tokenizer=tiktoken.get_encoding("cl100k_base")defadd(self,content:str,memory_type:str="short_term"):"""添加记忆"""item=MemoryItem(content=content,memory_type=memory_type,timestamp=time.time())self.buffer.append(item)self._trim_by_tokens()defget_context(self,max_items:Optional[int]=None)-str:"""获取上下文"""items=self.buffer[-max_items:]ifmax_itemselseself.bufferreturn"\n".join([item.contentforiteminitems])defget_all(self)-List[MemoryItem]:"""获取所有记忆"""returnself.buffer.copy()def_trim_by_tokens(self):"""按Token数量裁剪"""total=sum(len(self.tokenizer.encode(item.content))foriteminself.buffer)whiletotalself.max_tokensandlen(self.buffer)2:removed=self.buffer.pop(0)total-=len(self.tokenizer.encode(removed.content))defclear(self):"""清空短期记忆"""self.buffer.clear()classLongTermMemory:""" 长期记忆：持久化存储关键经验和知识 基于向量数据库实现语义检索 """def__init__(self,vector_store,embeddings):self.store=vector_store# Pinecone/Weaviate客户端self.embeddings=embeddings# OpenAI embeddings等asyncdefstore_experience(self,task:str,outcome:str,lesson:str,metadata:Optional[Dict]=None):""" 存储经验到长期记忆 """doc_content=f""" 任务:{task}结果:{outcome}经验教训:{lesson}"""doc={"pageContent":doc_content.strip(),"metadata":{"type":"experience","timestamp":time.time(),"task_type":metadata.get("task_type")ifmetadataelseNone,**(metadataor{})}}awaitself.store.add_documents([doc])# 同步到经验库awaitself._sync_to_experience_store(task,outcome,lesson)asyncdefrecall(self,query:str,k:int=5)-List[str]:""" 检索相关记忆 """results=awaitself.store.similarity_search(query,k=k)return[r.pageContentforrinresults]asyncdef_sync_to_experience_store(self,task:str,outcome:str,lesson:str):"""同步到结构化经验库"""# 提取关键模式passasyncdefconsolidate(self,session_summary:str):""" 记忆巩固：将重要短期记忆迁移到长期记忆 """# 识别重要信息important_patterns=awaitself._extract_important_patterns(session_summary)forpatterninimportant_patterns:awaitself.store_experience(task=pattern["task"],outcome=pattern["outcome"],lesson=pattern["lesson"])classHierarchicalMemory:""" 分层记忆管理器 协调短期记忆和长期记忆的交互 """def__init__(self,vector_store,embeddings,short_term_max_tokens:int=4000):self.short_term=ShortTermMemory(max_tokens=short_term_max_tokens)self.long_term=LongTermMemory(vector_store,embeddings)self.experience_store={}# 简化版经验库asyncdefremember(self,query:str,k:int=3)-List[str]:""" 统一检索接口：先查短期记忆，再查长期记忆 """# 1. 检查短期记忆short_results=self._search_short_term(query)# 2. 查长期记忆long_results=awaitself.long_term.recall(query,k=k)# 3. 合并去重combined=short_results+long_resultsreturnself._deduplicate(combined)[:k]def_search_short_term(self,query:str)-List[str]:"""短期记忆检索（简单关键词匹配）"""results=[]query_keywords=set(query.lower().split())foriteminself.short_term.get_all():item_keywords=set(item.content.lower().split())ifquery_keywordsitem_keywords:# 有交集results.append(item.content)returnresultsdefupdate(self,content:str):"""更新记忆"""self.short_term.add(content)asyncdefconsolidate_session(self,summary:str):"""会话结束时巩固记忆"""awaitself.long_term.consolidate(summary)2.4 规划模块（Planning Module）规划模块负责将复杂目标分解为可执行的子任务序列：# Python实现：基于CoT/ToT的规划模块fromtypingimportList,Optional,Dict,Any,CallablefromdataclassesimportdataclassfromenumimportEnumimportjsonclassTaskStatus(Enum):PENDING="pending"IN_PROGRESS="in_progress"COMPLETED="completed"FAILED="failed"BLOCKED="blocked"@dataclassclassSubTask:"""子任务定义"""id:strdescription:strstatus:TaskStatus=TaskStatus.PENDING dependencies:List[str]=None# 依赖的子任务IDtool_required:Optional[str]=Noneresult:Optional[Any]=Noneerror:Optional[str]=Nonedef__post_init__(self):ifself.dependenciesisNone:self.dependencies=[]classPlanningModule:""" 规划模块：任务分解与执行计划生成 支持多种规划策略： 1. Chain of Thought (CoT) - 链式思维 2. Tree of Thoughts (ToT) - 思维树 3. Subgoal Decomposition - 子目标分解 """def__init__(self,llm):self.llm=llmasyncdefcreate_plan(self,goal:str,strategy:str="cot",context:Optional[Dict]=None)-List[SubTask]:""" 创建执行计划 """ifstrategy=="cot":returnawaitself._plan_with_cot(goal,context)elifstrategy=="tot":returnawaitself._plan_with_tot(goal,context)elifstrategy=="simple":returnawaitself._plan_simple(goal)else:raiseValueError(f"Unknown strategy:{strategy}")asyncdef_plan_with_cot(self,goal:str,context:Optional[Dict])-List[SubTask]:""" Chain of Thought 规划 让LLM展示完整推理过程 """prompt=f""" 目标：{goal}{f"上下文信息：{json.dumps(context,ensure_ascii=False)}"ifcontextelse""}请按以下步骤思考并输出计划： 1. 分析目标的核心需求 2. 识别完成目标需要的步骤 3. 确定步骤之间的依赖关系 4. 输出结构化的任务列表 输出格式（JSON）： { { "reasoning": "你的推理过程", "tasks": [ { {"id": "1", "description": "任务1", "dependencies": []}}, { {"id": "2", "description": "任务2", "dependencies": ["1"]}} ] }} """response=awaitself.llm.agenerate([prompt])result=json.loads(response)tasks=[SubTask(id=t["id"],description=t["description"],dependencies=t.get("dependencies",[]))fortinresult["tasks"]]returntasksasyncdef_plan_with_tot(self,goal:str,context:Optional[Dict])-List[SubTask]:""" Tree of Thoughts 规划 探索多条执行路径，选择最优方案 """# 第一阶段：生成多个可能的计划prompt=f""" 目标：{goal}请生成3种不同的执行方案，每种方案从不同角度思考： 方案A：从XX角度... 方案B：从YY角度... 方案C：从ZZ角度... 对于每种方案，列出关键步骤。 """response=awaitself.llm.agenerate([prompt])# 第二阶段：评估选择最优方案evaluation_prompt=f""" 针对目标"