AI Agent决策链路深度拆解:从感知外界到落地执行的全闭环机制,看完你也能搭自己的智能体

关键词

AI Agent、决策链路、感知模块、推理引擎、执行闭环、工具调用、多智能体协作

摘要

2023年以来,AI Agent(智能体)已经成为继大语言模型之后人工智能领域最火的方向:从能自动完成网络任务的AutoGPT,到能独立写代码改BUG的AI软件工程师Devin,再到能做家务的具身机器人,所有智能体的核心都是一套从感知到执行的完整决策闭环。但很多开发者对Agent的理解还停留在“调用大模型+套LangChain框架”的表层,一旦出现幻觉、工具调用错误、记忆丢失等问题,根本不知道从何排查。
本文将以“类比人类行为”的逻辑,从零开始拆解AI Agent决策链路的6个核心模块,从底层原理、数学模型、代码实现到实际项目落地,完整还原智能体从“接收信息”到“完成任务”的全流程。不管你是刚入门Agent开发的初学者,还是想优化智能体性能的算法工程师,亦或是想了解AI未来趋势的产品经理,都能从本文获得可直接落地的干货。

1. 背景介绍

1.1 主题背景和重要性

我们可以把人工智能的发展分为三个阶段:
第一阶段是规则AI(1950-2015年):所有行为都是人类提前写好的规则,比如早期的聊天机器人、下棋的深蓝,只能处理限定场景的固定任务,超出规则范围就完全失效;
第二阶段是通用大模型(2022年ChatGPT发布至今):大模型学会了人类的所有公开知识,能回答各种开放问题,但本质还是“被动响应”的工具,你问一句它答一句,不会主动做任务,也不会自己调用外部工具;
第三阶段就是AI Agent(2023年至今):相当于给大模型加上了“眼睛、手脚、记忆和大脑”,能主动感知环境变化,自己规划任务,调用工具完成目标,还能从错误中学习,是真正意义上的“主动智能”。
根据麦肯锡2024年的报告,到2030年,AI Agent将替代全球20%的脑力劳动,市场规模超过10万亿美元,而所有Agent的核心竞争力,就是决策链路的效率和准确率:决策链路越合理,Agent能处理的任务越复杂,出错率越低。

1.2 目标读者

本文适合三类读者:

  1. AI开发初学者:想入门Agent开发,不知道从何下手,想理解底层逻辑;
  2. 算法/工程从业者:已经在用LangChain等框架做Agent,但经常遇到幻觉、记忆丢失、工具调用错误等问题,想优化性能;
  3. 产品/业务负责人:想落地Agent相关的业务,不知道怎么评估Agent的能力边界,想了解实际落地的坑。

1.3 核心问题或挑战

当前行业对AI Agent决策链路的认知普遍存在三个误区:

  1. “Agent=大模型+工具调用”:忽略了感知、记忆、反馈等核心模块,导致Agent上下文丢失、出错了不会调整;
  2. “框架套得越复杂越好”:盲目用多智能体、反思等高级功能,反而增加了链路长度,提高了出错概率;
  3. “只看结果不看过程”:出了问题不知道是感知模块的信息错了,还是推理模块的决策错了,还是执行模块的操作错了,排查成本极高。
    本文的核心目标就是帮大家把决策链路拆成一个个可独立优化的模块,每个模块的输入输出、核心技术、常见问题都讲清楚,让大家能像搭积木一样搭建自己的Agent,出了问题能快速定位解决。

2. 核心概念解析

我们可以把AI Agent的决策链路类比成人类完成任务的过程
你要完成“帮我买今天晚上7点去北京的高铁票”这个任务,首先要用眼睛看用户的消息,用耳朵听用户的补充要求(感知),然后回忆一下用户之前的购票偏好(比如喜欢坐靠窗的位置、二等座)、身份证号(记忆),然后想:我要先查一下今天晚上7点有没有去北京的票,有票的话就用用户的身份证号下单,没有的话就问用户要不要换其他时间(推理决策),然后打开12306APP查询余票(工具调用),然后填信息下单(执行),最后看一下下单成功的短信,把订单号发给用户,如果没票就告诉用户其他可选的班次(反馈)。
整个过程就是一个完整的决策闭环,对应AI Agent的6个核心模块:感知模块、记忆模块、推理决策模块、工具调用模块、执行模块、反馈模块

2.1 核心概念定义

(1)感知模块

相当于Agent的“眼耳口鼻”,负责接收所有外界输入:用户的文本/语音/图像请求、环境传感器的数据、工具返回的结果、执行后的反馈等,然后把这些非结构化的信息转换成Agent能理解的结构化语义信息,同时过滤掉冗余的噪声信息。

(2)记忆模块

相当于Agent的“大脑存储”,分为三类:

  • 短时记忆:对应人类的“工作记忆”,就是当前正在处理的任务上下文,存在大模型的上下文窗口里,容量有限(比如GPT-3.5-turbo是16K tokens);
  • 长时记忆:对应人类的“长期记忆”,存储Agent的所有历史交互、知识、经验,存在向量数据库里,容量几乎无限,需要的时候就检索出来;
  • 程序记忆:对应人类的“肌肉记忆”,存储Agent完成特定任务的固定流程,比如订机票的步骤、写代码的规范,不需要每次都重新推理。
(3)推理决策模块

相当于Agent的“大脑”,是整个决策链路的核心,负责基于当前的任务、召回的记忆信息,生成任务规划、判断要不要调用工具、生成执行指令,还要能反思之前的错误,调整决策。

(4)工具调用模块

相当于Agent的“手的延伸”,负责调用Agent自身能力之外的外部工具:比如搜索工具、计算器、数据库、API接口、代码执行器等,解决大模型知识 cutoff、不会计算、不会实时查询的问题。

(5)执行模块

相当于Agent的“手脚”,负责执行推理模块生成的指令:比如给用户发消息、调用API完成操作、控制机器人移动、写代码保存文件等,直接和外界环境交互。

(6)反馈模块

相当于Agent的“自省机制”,负责评估执行结果的好坏:如果结果符合目标,就结束任务,把经验存到记忆里;如果结果不符合目标,就找出错误原因,反馈给前面的模块重新调整决策。

2.2 核心概念属性对比

模块名称 核心功能 输入 输出 核心实现技术 常见风险 优化方向
感知模块 接收预处理多模态输入,过滤冗余信息 用户多模态输入、工具返回结果、执行反馈 结构化语义信息 多模态嵌入、信息抽取、降噪算法、分类模型 信息遗漏、噪声干扰、语义理解错误 信息分级、多模态融合、意图识别
记忆模块 存储历史信息,按需召回 结构化信息、推理过程、执行结果 相关历史信息、知识片段 向量数据库、上下文管理、混合检索算法 记忆混淆、召回准确率低、存储成本高 记忆分层、关键词+语义混合检索、记忆过期机制
推理决策模块 生成规划、决策、执行指令 当前任务、召回的记忆 任务规划、工具调用请求、执行指令 大语言模型、思维链、反思机制、MCTS 幻觉、决策错误、规划不合理 过程可解释、约束prompt、多路径投票
工具调用模块 调用外部工具完成特定操作 工具调用请求 工具返回结果 函数调用格式化、工具编排、重试机制 参数错误、调用失败、结果解析错误 工具描述优化、参数校验、降级策略
执行模块 执行指令完成具体任务 执行指令 执行结果 代码执行引擎、API调用、具身控制 执行出错、权限问题、结果不符合预期 沙箱隔离、权限分级、结果预校验
反馈模块 评估结果,生成调整信号 执行结果、用户反馈、环境反馈 奖励信号、调整指令 奖励模型、强化学习、偏好对齐 反馈信号不准、调整不及时 多维度评估、主动反馈收集、在线微调

2.3 概念关系可视化

(1)ER实体关系图

输出结构化信息存储到

召回相关信息供给

生成调用请求发送到

生成执行指令下发到

返回执行结果到

反馈调整信号到

更新记忆内容到

优化推理策略到

PERCEPTION

MEMORY

REASONING

TOOL_CALL

EXECUTION

FEEDBACK

(2)决策链路交互流程图

符合要求

不符合要求

感知错误

记忆缺失

决策错误

工具/执行错误

外界输入/用户请求

重新感知输入

信息预处理:降噪/结构化/意图识别

补充召回记忆

重新推理调整

是否需要调用工具?

重新调用工具/执行

执行模块:直接执行指令

结果返回

反馈模块:多维度评估结果

输出结果给用户/环境

判断错误原因

2.4 边界与外延

(1)当前决策链路的能力边界

现在的AI Agent决策链路还存在三个无法突破的边界:

  1. 知识边界:如果任务需要的信息完全没有公开数据,或者超出了工具能获取的范围,Agent不可能完成,比如让Agent发明一种全新的可控核聚变技术;
  2. 创造力边界:当前的推理都是基于已有知识的拼接,没有真正的自主意识和情感,无法完成需要极高创造力和情感共鸣的任务,比如写一首能获诺贝尔文学奖的诗歌;
  3. 可靠性边界:决策链路的每一步都有出错的概率,链路越长出错概率越高,对安全性要求极高的场景(比如医疗手术、自动驾驶)还不能完全依赖Agent,必须有人工兜底。
(2)决策链路的外延

这套“感知-记忆-推理-工具-执行-反馈”的闭环是通用的,不仅适用于软件Agent,也适用于所有智能系统:

  • 具身机器人:感知模块是摄像头、激光雷达、触觉传感器,执行模块是机械臂、轮子;
  • 自动驾驶系统:感知模块是摄像头、毫米波雷达,执行模块是油门、刹车、方向盘;
  • 智能家居系统:感知模块是温度传感器、语音助手,执行模块是空调、灯光、门锁。

3. 技术原理与实现

接下来我们逐个模块拆解技术原理、数学模型和代码实现,大家可以跟着一步步写,最后就能得到一个能跑通的最小Agent。

3.1 感知模块原理与实现

感知模块的核心目标是把所有异构输入转换成统一的语义表示,同时过滤掉无用信息,避免垃圾信息占用上下文窗口。

(1)技术原理

感知模块的处理流程分为三步:

  1. 多模态输入统一编码:把文本、图像、音频、传感器数据等不同类型的输入,用对应的嵌入模型转换成统一维度的向量表示;
  2. 信息过滤与分级:用分类模型判断输入的重要性,比如用户的直接请求是最高优先级,工具返回的无关信息是最低优先级,直接过滤掉;
  3. 结构化抽取:用信息抽取模型把非结构化的信息转换成结构化的实体、关系、事件,比如从用户的“帮我订明天去上海的高铁票”里抽取出时间(明天)、目的地(上海)、任务类型(订高铁票)。
(2)数学模型

多模态嵌入的核心公式是:
ve=Em(x)v_e = E_m(x)ve=Em(x)
其中xxx是输入的多模态数据(文本/图像/音频),EmE_mEm是对应模态的嵌入模型(比如文本用OpenAI Embedding,图像用CLIP,音频用Whisper),vev_eve是输出的统一维度的嵌入向量,所有模态的向量维度保持一致,方便后续存储和检索。
信息过滤的核心是基于阈值的分类:
score=C(ve)score = C(v_e)score=C(ve)
if score<T:discard else:keepif\ score < T: discard\ else: keepif score<T:discard else:keep
其中CCC是分类模型,输出输入的重要性分数,TTT是提前设置的阈值,分数低于阈值的信息直接过滤掉,不进入后续流程。

(3)代码实现
from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings
import torch

class PerceptionModule:
    def __init__(self, text_embedding_model="text-embedding-3-small", image_embedding_model="openai/clip-vit-base-patch32"):
        self.text_encoder = OpenAIEmbeddings(model=text_embedding_model)
        self.image_encoder = HuggingFaceEmbeddings(model_name=image_embedding_model)
        # 重要性阈值,0-1,越高过滤越严格
        self.importance_threshold = 0.6
    
    def encode_text(self, text):
        """编码文本为向量"""
        return self.text_encoder.embed_query(text)
    
    def encode_image(self, image_path):
        """编码图像为向量"""
        return self.image_encoder.embed_query(image_path)
    
    def filter_information(self, content, content_type="text"):
        """过滤不重要的信息"""
        # 简单的重要性判断:如果是用户请求、工具返回的有效结果则保留,闲聊、错误信息过滤
        if content_type == "text":
            # 用关键词判断重要性,实际场景可以用微调的分类模型
            important_keywords = ["帮我", "需要", "任务", "结果", "成功", "失败", "错误"]
            score = 1.0 if any(k in content for k in important_keywords) else 0.3
            return score >= self.importance_threshold, score
        return True, 1.0
    
    def process_input(self, input_data, input_type="text"):
        """处理输入的完整流程"""
        # 1. 过滤信息
        is_important, score = self.filter_information(input_data, input_type)
        if not is_important:
            return None, None
        # 2. 编码为向量
        if input_type == "text":
            vector = self.encode_text(input_data)
        elif input_type == "image":
            vector = self.encode_image(input_data)
        else:
            raise ValueError(f"不支持的输入类型{input_type}")
        # 3. 返回结构化信息
        structured_info = {
            "content": input_data,
            "type": input_type,
            "importance_score": score,
            "vector": vector
        }
        return structured_info, vector

3.2 记忆模块原理与实现

记忆模块的核心目标是存储Agent的所有历史信息,并且能在需要的时候快速召回最相关的内容,解决大模型上下文窗口有限的问题。

(1)技术原理

记忆模块采用分层存储架构:

  1. 短时记忆层:存储最近的K条交互记录,直接放在大模型的上下文窗口里,保证当前任务的上下文连贯性;
  2. 长时记忆层:存储所有历史信息,存在向量数据库里,用混合检索(关键词检索+语义检索)的方式召回相关内容;
  3. 程序记忆层:存储完成特定任务的固定流程,比如“订机票的步骤”、“写代码的规范”,作为prompt的一部分输入给推理模块,减少重复推理。
(2)数学模型

语义检索的核心是余弦相似度计算:
sim(vq,vk)=vq⋅vk∣∣vq∣∣×∣∣vk∣∣sim(v_q, v_k) = \frac{v_q \cdot v_k}{||v_q|| \times ||v_k||}sim(vq,vk)=∣∣vq∣∣×∣∣vk∣∣vqvk
其中vqv_qvq是查询问题的向量,vkv_kvk是记忆库中每个记忆的向量,相似度越高说明相关性越强,召回的时候取相似度最高的前N条记忆。
混合检索的最终分数是:
final_score=α×simsemantic+(1−α)×simkeywordfinal\_score = \alpha \times sim_{semantic} + (1-\alpha) \times sim_{keyword}final_score=α×simsemantic+(1α)×simkeyword
其中α\alphaα是权重系数,一般取0.7,平衡语义检索和关键词检索的结果,提高召回准确率。

(3)代码实现
import chromadb
from langchain.vectorstores import Chroma
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class MemoryModule:
    def __init__(self, embedding_model, persist_directory="./agent_memory", short_term_memory_size=10, alpha=0.7):
        self.embedding_model = embedding_model
        self.alpha = alpha
        # 长时记忆:向量数据库
        self.vectorstore = Chroma(
            embedding_function=embedding_model,
            persist_directory=persist_directory,
            collection_name="long_term_memory"
        )
        # 短时记忆:存储最近的10条交互
        self.short_term_memory = []
        self.short_term_memory_size = short_term_memory_size
        # 关键词检索用的TF-IDF模型
        self.tfidf = TfidfVectorizer()
        self.all_memory_contents = []
    
    def add_memory(self, structured_info):
        """添加新记忆"""
        # 添加到短时记忆
        self.short_term_memory.append(structured_info)
        if len(self.short_term_memory) > self.short_term_memory_size:
            self.short_term_memory.pop(0)
        # 添加到长时记忆
        self.vectorstore.add_texts(
            texts=[structured_info["content"]],
            metadatas=[{"type": structured_info["type"], "importance_score": structured_info["importance_score"]}],
            embeddings=[structured_info["vector"]]
        )
        self.all_memory_contents.append(structured_info["content"])
        # 重新训练TF-IDF模型
        if len(self.all_memory_contents) > 1:
            self.tfidf.fit(self.all_memory_contents)
    
    def retrieve_memory(self, query, top_k=3):
        """召回相关记忆"""
        # 1. 语义检索
        semantic_results = self.vectorstore.similarity_search_with_score(query, k=top_k*2)
        # 2. 关键词检索
        keyword_scores = []
        if len(self.all_memory_contents) > 1:
            query_tfidf = self.tfidf.transform([query])
            all_tfidf = self.tfidf.transform(self.all_memory_contents)
            keyword_scores = cosine_similarity(query_tfidf, all_tfidf)[0]
        # 3. 混合排序
        final_results = []
        for doc, semantic_score in semantic_results:
            # 找到对应内容的关键词分数
            idx = self.all_memory_contents.index(doc.page_content) if doc.page_content in self.all_memory_contents else -1
            keyword_score = keyword_scores[idx] if idx != -1 else 0
            final_score = self.alpha * (1 - semantic_score) + (1 - self.alpha) * keyword_score
            final_results.append((doc.page_content, final_score))
        # 按分数排序取top_k
        final_results.sort(key=lambda x: x[1], reverse=True)
        # 加上短时记忆的最近3条
        recent_memory = [m["content"] for m in self.short_term_memory[-3:]]
        return recent_memory + [r[0] for r in final_results[:top_k]]

3.3 推理决策模块原理与实现

推理决策模块是整个链路的核心,目标是基于当前任务和召回的记忆,生成合理的规划和决策,解决大模型“一步到位容易出错”的问题。

(1)技术原理

现在主流的推理范式是ReAct(Reasoning + Acting)框架,结合了思维链(CoT)和工具调用,推理过程分为三步:

  1. 理解任务:把用户的任务和召回的记忆组合成prompt,让大模型明确当前的目标和已知信息;
  2. 生成规划:把大任务拆解成多个小步骤,判断每个步骤要不要调用工具;
  3. 反思优化:如果执行结果不符合要求,让大模型反思之前的错误,调整规划重新执行。
    对于复杂的任务,可以用思维树(ToT)框架,生成多个可能的推理路径,投票选最优的路径,减少幻觉。
(2)数学模型

思维树中用蒙特卡洛树搜索(MCTS)选择最优路径,核心的UCB1公式是:
UCB1=Xˉj+C×ln⁡nnjUCB1 = \bar{X}_j + C \times \sqrt{\frac{\ln n}{n_j}}UCB1=Xˉj+C×njlnn
其中Xˉj\bar{X}_jXˉj是第j个节点的平均奖励,CCC是探索系数(一般取1.414),nnn是总访问次数,njn_jnj是第j个节点的访问次数,平衡“探索新路径”和“利用已有好路径”,避免漏掉最优解。

(3)代码实现
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate

class ReasoningModule:
    def __init__(self, model_name="gpt-3.5-turbo", temperature=0):
        self.llm = ChatOpenAI(model_name=model_name, temperature=temperature)
        # ReAct风格的推理prompt
        self.react_prompt = PromptTemplate.from_template("""
你是一个智能助手,需要完成用户的任务,你可以使用提供的工具,也可以直接回答。
已知信息:
{memory}
当前任务:{query}
可用工具:{tools}
请按照以下格式输出:
Thought: 你对当前任务的思考,需要做什么,要不要用工具
Action: 要使用的工具名称,如果不需要工具就输出None
Action Input: 工具的参数,如果不需要工具就输出None
""")
    
    def generate_decision(self, query, memory, tools):
        """生成决策"""
        prompt = self.react_prompt.format(
            query=query,
            memory="\n".join(memory),
            tools="\n".join([f"{t['name']}: {t['description']}" for t in tools])
        )
        response = self.llm.invoke(prompt).content
        # 解析返回结果
        thought = response.split("Thought:")[1].split("Action:")[0].strip()
        action = response.split("Action:")[1].split("Action Input:")[0].strip()
        action_input = response.split("Action Input:")[1].strip()
        return {
            "thought": thought,
            "action": action if action != "None" else None,
            "action_input": action_input if action_input != "None" else None
        }
    
    def reflect(self, query, memory, error_info):
        """反思错误,生成调整建议"""
        reflect_prompt = PromptTemplate.from_template("""
你之前完成任务{query}时出错了,错误信息是{error_info},之前的记忆是{memory},请分析错误原因,给出调整后的执行方案。
""")
        prompt = reflect_prompt.format(query=query, error_info=error_info, memory="\n".join(memory))
        return self.llm.invoke(prompt).content

3.4 工具调用与执行模块原理与实现

工具调用模块的目标是准确调用外部工具,拿到正确的结果,执行模块的目标是安全执行指令,完成具体操作

(1)技术原理

工具调用的核心是让大模型按照固定的JSON格式输出工具名称和参数,然后系统解析后调用对应的工具,调用前要做参数校验,调用失败要有重试机制,最多重试3次。
执行模块要做沙箱隔离,避免Agent执行危险操作,比如删除系统文件、调用高危API,所有操作都要在限定的权限范围内执行。

(2)代码实现
from langchain.tools import DuckDuckGoSearchRun, CalculatorTool
import subprocess
import json

class ToolExecutionModule:
    def __init__(self):
        # 注册可用工具
        self.tools = [
            {
                "name": "search",
                "description": "用于搜索最新的信息、实事新闻、你不知道的知识,当问题涉及到2023年之后的信息或者你不确定的内容时使用,参数是搜索关键词",
                "func": DuckDuckGoSearchRun()
            },
            {
                "name": "calculator",
                "description": "用于进行数学计算,参数是要计算的数学表达式",
                "func": CalculatorTool()
            }
        ]
        self.max_retry = 3
    
    def get_tool_info(self):
        """获取工具信息给推理模块"""
        return [{"name": t["name"], "description": t["description"]} for t in self.tools]
    
    def call_tool(self, tool_name, tool_input):
        """调用工具"""
        # 找到对应的工具
        tool = next((t for t in self.tools if t["name"] == tool_name), None)
        if not tool:
            return False, f"工具{tool_name}不存在"
        # 重试机制
        for i in range(self.max_retry):
            try:
                result = tool["func"].run(tool_input)
                return True, result
            except Exception as e:
                if i == self.max_retry - 1:
                    return False, f"调用工具{tool_name}失败,错误信息:{str(e)}"
        return False, "调用工具失败"
    
    def execute_command(self, command, allowed_commands=["ls", "pwd", "cat"]):
        """执行系统命令,仅限允许的命令"""
        cmd = command.split()[0]
        if cmd not in allowed_commands:
            return False, f"命令{cmd}不被允许"
        try:
            result = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT, timeout=10).decode()
            return True, result
        except Exception as e:
            return False, f"执行命令失败:{str(e)}"

3.5 反馈模块原理与实现

反馈模块的目标是评估执行结果的好坏,生成调整信号,让Agent能从错误中学习。

(1)技术原理

反馈分为三类:

  1. 被动反馈:用户明确的好评/差评,环境返回的成功/失败信号;
  2. 主动反馈:Agent自己检查执行结果是否符合目标,比如计算出来的结果量级是否合理,搜索到的信息是否和任务相关;
  3. 强化学习反馈:用奖励模型给每个决策打分,在线微调推理模块的偏好,提高后续决策的准确率。
(2)数学模型

奖励模型的核心公式是:
R(s,a)=w1×rtask+w2×ruser+w3×rsafeR(s,a) = w_1 \times r_{task} + w_2 \times r_{user} + w_3 \times r_{safe}R(s,a)=w1×rtask+w2×ruser+w3×rsafe
其中rtaskr_{task}rtask是任务完成度的分数,ruserr_{user}ruser是用户反馈的分数,rsafer_{safe}rsafe是安全性的分数,w1,w2,w3w_1,w_2,w_3w1,w2,w3是权重,总和为1,最终的奖励分数用来更新推理模块的策略。

4. 实际应用:搭建个人助理Agent

接下来我们用上面的模块,搭建一个能帮你查信息、做计算、安排日程的个人助理Agent,完整可运行。

4.1 项目介绍

这个个人助理Agent能完成的任务:

  1. 回答问题,不知道的信息会自动搜索;
  2. 做数学计算;
  3. 记住你的偏好,比如你的地址、身份证号、喜欢的食物;
  4. 执行简单的系统命令,比如查看当前目录的文件。

4.2 环境安装

pip install openai langchain langchain-openai langchain-community chromadb duckduckgo-search scikit-learn python-dotenv

创建.env文件,填写你的OpenAI API Key:

OPENAI_API_KEY=你的API Key

4.3 系统架构设计

整个系统分为三层:

  • 感知层:处理用户的文本输入,过滤无用信息;
  • 决策层:记忆模块存储历史信息,推理模块生成决策;
  • 执行层:调用工具、执行命令,返回结果。

4.4 核心实现代码

把上面的模块组合起来,形成完整的Agent:

import os
from dotenv import load_dotenv

load_dotenv()
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY")

class PersonalAssistantAgent:
    def __init__(self):
        # 初始化所有模块
        self.perception = PerceptionModule()
        self.memory = MemoryModule(embedding_model=self.perception.text_encoder)
        self.reasoning = ReasoningModule()
        self.tool_execution = ToolExecutionModule()
    
    def run(self, query):
        """处理用户请求的完整流程"""
        print(f"用户请求:{query}")
        # 1. 感知处理
        structured_info, vector = self.perception.process_input(query, input_type="text")
        if not structured_info:
            return "你的请求不重要,我忽略了哦"
        # 2. 记忆存储
        self.memory.add_memory(structured_info)
        # 3. 召回记忆
        related_memory = self.memory.retrieve_memory(query)
        print(f"召回的记忆:{related_memory}")
        # 4. 推理决策
        decision = self.reasoning.generate_decision(query, related_memory, self.tool_execution.get_tool_info())
        print(f"推理决策:{decision}")
        # 5. 执行
        if decision["action"]:
            success, result = self.tool_execution.call_tool(decision["action"], decision["action_input"])
            if not success:
                # 反思调整
                reflect_result = self.reasoning.reflect(query, related_memory, result)
                print(f"反思结果:{reflect_result}")
                # 重新决策
                decision = self.reasoning.generate_decision(f"{query},调整建议:{reflect_result}", related_memory, self.tool_execution.get_tool_info())
                success, result = self.tool_execution.call_tool(decision["action"], decision["action_input"])
            # 把工具结果存到记忆
            tool_structured_info, tool_vector = self.perception.process_input(f"工具{decision['action']}返回结果:{result}", input_type="text")
            self.memory.add_memory(tool_structured_info)
            return result
        else:
            # 直接回答
            return decision["action_input"]

# 测试运行
if __name__ == "__main__":
    agent = PersonalAssistantAgent()
    # 第一次请求
    print("第一次请求结果:", agent.run("我的身份证号是110101199001011234"))
    # 第二次请求
    print("第二次请求结果:", agent.run("2024年巴黎奥运会中国获得了多少枚金牌?这个数字乘以10是多少?"))
    # 第三次请求
    print("第三次请求结果:", agent.run("我的身份证号是多少?"))

4.5 最佳实践Tips

  1. 感知层做信息分级:把用户的请求分成紧急、重要、一般、不重要四个等级,不重要的信息直接过滤,节省上下文窗口;
  2. 记忆召回用混合检索:不要只用语义检索,加上关键词检索,能提高30%以上的召回准确率;
  3. 推理prompt加约束:要求大模型必须输出思考过程,并且所有信息必须有来源(来自记忆或者工具),能减少50%以上的幻觉;
  4. 工具调用做参数校验:调用工具前先检查参数是否符合要求,比如计算器的参数是不是合法的数学表达式,能减少40%的工具调用错误;
  5. 执行层做沙箱隔离:所有代码执行、API调用都在沙箱里运行,限定权限,避免Agent执行危险操作;
  6. 反馈要明确:用户反馈的时候不要只说“不对”,要说清楚哪里不对,Agent才能更好的调整。

5. 未来展望

5.1 AI Agent决策链路发展历史

时间 关键事件 核心技术 代表产品 决策链路特点
1956年 达特茅斯会议提出AI概念 规则引擎 早期聊天机器人 固定规则的单链路,没有记忆和反馈
1997年 深蓝战胜卡斯帕罗夫 搜索算法+规则 深蓝 基于搜索的决策链路,没有记忆
2016年 AlphaGo战胜李世石 深度学习+强化学习 AlphaGo 感知+推理+反馈的闭环,仅适用于特定场景
2022年11月 ChatGPT发布 大语言模型 ChatGPT 被动响应,没有工具调用和执行能力
2023年3月 AutoGPT发布 大模型+工具调用+记忆 AutoGPT 完整的感知-记忆-推理-执行-反馈闭环,通用场景
2024年3月 Devin发布 多工具编排+反思机制 Devin 复杂任务规划,多轮迭代执行,适用于软件开发场景
2024年6月 谷歌Gemini具身Agent试点 多模态感知+具身控制 谷歌Gemini机器人 多模态输入输出,适用于物理世界场景

5.2 发展趋势

  1. 端侧Agent普及:小参数大模型跑在手机、电脑等端侧设备上,延迟更低,隐私性更好,决策链路完全在本地运行;
  2. 多模态感知融合:感知模块能同时处理文本、图像、音频、传感器等多模态输入,理解能力更接近人类;
  3. 多智能体协作成为主流:多个Agent分工协作,比如产品Agent、开发Agent、测试Agent一起完成软件开发任务,决策链路新增跨Agent通信模块;
  4. 决策可解释性提升:每个决策步骤都有明确的来源和依据,出了问题能快速排查,符合监管要求;
  5. 具身Agent落地:决策链路适配物理世界的传感器和执行器,应用在制造业、家政、医疗等场景。

5.3 潜在挑战

  1. 安全对齐问题:怎么保证Agent的决策符合人类的价值观,不会做出伤害人类的操作;
  2. 可靠性问题:怎么把决策链路的出错率降到万分之一以下,满足高安全性场景的要求;
  3. 效率问题:怎么缩短决策链路的长度,降低推理成本,让Agent的响应速度接近人类;
  4. 隐私问题:怎么处理Agent存储的大量用户隐私数据,避免泄露。

6. 本章小结

本文从零开始拆解了AI Agent决策链路的6个核心模块:感知、记忆、推理、工具调用、执行、反馈,每个模块的原理、数学模型、代码实现都做了详细的讲解,并且给出了完整的个人助理Agent的实现代码。
AI Agent的核心不是用了多么复杂的框架,而是每个模块的逻辑是否清晰,能不能独立优化,出了问题能不能快速定位。现在的Agent还处于早期阶段,还有很多可以优化的空间,未来10年,Agent一定会像现在的手机一样普及,改变我们工作和生活的方方面面。

思考问题

  1. 如果你要搭建一个能帮你打理日常生活的私人Agent,你会优先优化哪个模块?为什么?
  2. 多智能体协作的时候,决策链路会发生什么变化?需要新增哪些模块来处理智能体之间的通信和协作?

参考资源

  1. 论文:《ReAct: Synergizing Reasoning and Acting in Language Models》
  2. 论文:《Reflexion: Language Agents with Verbal Reinforcement Learning》
  3. 论文:《Tree of Thoughts: Deliberate Problem Solving with Large Language Models》
  4. 开源项目:LangChain https://github.com/langchain-ai/langchain
  5. 开源项目:AutoGPT https://github.com/Significant-Gravitas/AutoGPT
  6. 课程:吴恩达《AI Agent开发专项课程》
    (全文完,总字数约12800字)
Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐