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的核心价值

二、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,metadata

2.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)whiletotal>self.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_keywords&item_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""" 针对目标"