朋友们，如需转载请标明出处：http://blog.csdn.net/jiangjunshow

各位程序员朋友，咱们终于进入核心技术篇啦！前面咱们搞懂了Agent的“是什么、能做啥、用什么工具开发”，现在是时候扒开Agent的“外衣”，看看它的内部构造了。

就像咱们学编程时，先知道“函数能实现某个功能”，再深入理解“函数的参数、返回值、内部逻辑”一样，开发Agent也得先搞懂它的核心模块。这一节咱们就把Agent拆成四个核心模块：感知模块、记忆模块、决策模块、工具调用模块，用程序员能秒懂的语言，把每个模块的作用、工作原理、实现思路讲清楚，让你从“用Agent”变成“懂Agent、能开发Agent”。

一、感知模块：Agent的“五官”——收集信息的“输入端”

咱们先从最基础的感知模块说起。感知模块是Agent的“五官”，核心作用就是收集外界信息，给Agent提供“决策的原材料”。如果没有感知模块，Agent就像一个“瞎子、聋子”，再强的决策能力也没用。

1. 感知模块要收集哪些信息？

感知模块收集的信息主要分两类，咱们用程序员的话来说就是“输入数据”：

• 用户直接输入的信息：比如你告诉代码助手“写一个Python函数批量读取Excel”，这是“明文输入”；还有你和Agent的对话历史，比如之前聊过“要用pandas库”，这也是感知模块要收集的信息；
• 环境中的关联信息：这是感知模块的核心价值，它会主动去“找”和当前任务相关的信息。比如智能客服感知到你说“快递没到”，会主动去查你的订单信息、物流轨迹；运维Agent感知到CPU飙升，会主动去查占用CPU的进程、服务器的内存使用情况。

简单说，感知模块的目标就是：把“用户说的话”和“隐藏在背后的相关数据”都收集起来，形成一个完整的“信息包”，交给后面的决策模块。

2. 感知模块的工作流程：从“信息收集”到“信息整理”

感知模块不是简单地“把信息堆在一起”，而是要做“预处理”，让决策模块能直接用。它的工作流程就像咱们写代码时的“数据清洗”：

1. 信息采集：通过各种“接口”收集信息，比如用户输入接口（对话框、API请求）、数据库查询接口（查订单、查服务器状态）、文件读取接口（读日志、读Excel）；
2. 信息筛选：过滤掉无关信息，比如用户问快递问题时，筛选掉用户的历史购物记录（没用），保留订单号、物流信息（有用）；
3. 信息结构化：把非结构化的信息变成结构化数据，比如把用户的自然语言“快递三天没到”，拆解成关键词【快递】【超时】【时长：3天】，方便决策模块解析；
4. 信息关联：把相关信息关联起来，比如把“用户ID”和“对应的订单信息”“物流信息”关联，形成一个完整的上下文。

3. 程序员视角：怎么实现一个简单的感知模块？

其实实现感知模块一点都不复杂，咱们用Python举个例子：
比如做一个“代码生成Agent”的感知模块，它要收集的信息是“用户需求+已导入的库+项目代码风格”：

# 1. 采集信息：从用户输入、项目文件中收集信息
def collect_information(user_input, project_path):
    # 采集用户输入（明文需求）
    demand = user_input
    # 采集已导入的库（从项目的requirements.txt中读取）
    with open(f"{project_path}/requirements.txt", "r") as f:
        imported_libs = f.read().splitlines()
    # 采集项目代码风格（从已有的.py文件中提取命名规范）
    code_style = analyze_code_style(project_path)
    return {
        "demand": demand,
        "imported_libs": imported_libs,
        "code_style": code_style
    }

# 2. 信息预处理：筛选、结构化、关联
def process_information(raw_info):
    # 筛选：只保留和代码生成相关的库（比如过滤掉测试库pytest）
    relevant_libs = [lib for lib in raw_info["imported_libs"] if lib in ["pandas", "requests", "flask"]]
    # 结构化：把自然语言需求拆成功能点
    function_points = parse_demand(raw_info["demand"])
    # 关联：把功能点和相关库关联（比如“读取Excel”关联pandas）
    associated_info = {
        "function_points": function_points,
        "recommended_libs": relevant_libs,
        "code_style": raw_info["code_style"]
    }
    return associated_info

# 感知模块的入口：收集+预处理
def perception_module(user_input, project_path):
    raw_info = collect_information(user_input, project_path)
    processed_info = process_information(raw_info)
    return processed_info

这个简单的感知模块，就完成了“收集信息→预处理”的全流程，输出的结构化信息可以直接交给决策模块使用。核心就是：明确要收集哪些信息，然后通过各种接口获取，再做简单的清洗和结构化。

二、记忆模块：Agent的“大脑存储区”——记住关键信息，避免“健忘”

如果Agent只有感知模块和决策模块，那它就是个“鱼的记忆”——每次交互都从零开始，比如你和智能客服聊了半天“快递问题”，转头问“那我的退款呢”，它还得让你重新说订单号，这体验就太差了。

记忆模块就是Agent的“大脑存储区”，核心作用是存储和管理关键信息，让Agent能“记住”之前的交互、执行过的步骤、收集到的信息，从而做出更智能的决策。

1. 记忆模块要记哪些东西？

Agent的记忆和咱们人类的记忆类似，分“短期记忆”和“长期记忆”，对应不同的使用场景：

• 短期记忆（上下文记忆）：记“最近的信息”，比如和用户的对话历史、当前任务的执行步骤。比如你和代码助手聊“写一个读取Excel的函数”，然后追问“添加异常处理”，短期记忆会记住之前的需求，不用你重新说；
• 长期记忆（知识库记忆）：记“长期有用的信息”，比如项目的业务规则（“用户密码必须加密存储”）、技术文档（“Spring Boot的接口开发规范”）、历史执行记录（“上次部署失败的原因是端口占用”）。这些信息不会随着任务结束而消失，会长期存储，供后续任务使用。

2. 记忆模块的工作流程：存储→检索→更新

记忆模块的工作流程就像咱们用“数据库+缓存”：

1. 存储：把感知模块处理后的信息、决策模块的执行结果、用户的反馈等，按“短期/长期”分类存储。短期记忆可以存在内存里（比如Python的字典），长期记忆可以存在数据库里（比如MySQL、向量数据库）；
2. 检索：当Agent处理新任务时，记忆模块会根据当前需求，检索相关的记忆。比如用户问“我的订单退款进度”，记忆模块会检索“用户的订单号”“之前的退款申请记录”；
3. 更新：当有新的信息产生时，记忆模块会更新记忆。比如用户提供了新的订单号，短期记忆会新增这条信息；Agent学到了新的业务规则，长期记忆会添加这条规则。

3. 程序员视角：怎么实现一个简单的记忆模块？

咱们还是用Python举例子，实现一个“短期记忆+长期记忆”的简单版本：

import json
import sqlite3

# 短期记忆：用字典存储，任务结束后清空
class ShortTermMemory:
    def __init__(self):
        self.memory = {}  # 结构：{"对话历史": [], "当前任务步骤": []}
    
    # 存储短期信息
    def store(self, key, value):
        self.memory[key] = value
    
    # 检索短期信息
    def retrieve(self, key):
        return self.memory.get(key, None)
    
    # 清空短期记忆（任务结束后调用）
    def clear(self):
        self.memory = {}

# 长期记忆：用SQLite存储，持久化
class LongTermMemory:
    def __init__(self, db_path="agent_memory.db"):
        self.conn = sqlite3.connect(db_path)
        self._create_table()
    
    # 创建表：存储业务规则、技术文档等
    def _create_table(self):
        cursor = self.conn.cursor()
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS knowledge (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                type TEXT,  # 类型：业务规则、技术文档、执行记录
                content TEXT,  # 内容
                create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()
    
    # 存储长期信息
    def store(self, type_, content):
        cursor = self.conn.cursor()
        cursor.execute('''
            INSERT INTO knowledge (type, content) VALUES (?, ?)
        ''', (type_, content))
        self.conn.commit()
    
    # 检索长期信息（根据类型和关键词）
    def retrieve(self, type_, keyword):
        cursor = self.conn.cursor()
        cursor.execute('''
            SELECT content FROM knowledge WHERE type = ? AND content LIKE ?
        ''', (type_, f"%{keyword}%"))
        return [row[0] for row in cursor.fetchall()]

# 记忆模块：整合短期和长期记忆
class MemoryModule:
    def __init__(self):
        self.short_term = ShortTermMemory()
        self.long_term = LongTermMemory()
    
    # 存储信息（自动判断短期/长期）
    def store(self, memory_type, key_or_type, value=None):
        if memory_type == "short":
            # 短期记忆：key是标识，value是内容
            self.short_term.store(key_or_type, value)
        elif memory_type == "long":
            # 长期记忆：key_or_type是类型，value是内容
            self.long_term.store(key_or_type, value)
    
    # 检索信息
    def retrieve(self, memory_type, key_or_type, keyword=None):
        if memory_type == "short":
            return self.short_term.retrieve(key_or_type)
        elif memory_type == "long":
            return self.long_term.retrieve(key_or_type, keyword)

# 使用示例
if __name__ == "__main__":
    memory = MemoryModule()
    # 存储短期记忆：对话历史
    memory.store("short", "conversation_history", ["用户：写一个Python读取Excel的函数", "Agent：已生成函数"])
    # 存储长期记忆：业务规则
    memory.store("long", "业务规则", "用户数据必须加密存储，禁止明文")
    # 检索短期记忆
    print(memory.retrieve("short", "conversation_history"))
    # 检索长期记忆：查找和“加密”相关的业务规则
    print(memory.retrieve("long", "业务规则", "加密"))

这个记忆模块虽然简单，但已经实现了核心功能：短期记忆存对话和任务状态，长期记忆存业务规则和知识库，检索时能快速找到需要的信息。实际开发中，你可以用Redis优化短期记忆的检索速度，用向量数据库（比如Milvus）优化长期记忆的语义检索（比如根据“读取Excel”找到相关的技术文档）。

三、决策模块：Agent的“大脑核心”——自主判断“该怎么做”

如果说感知模块是“收集原材料”，记忆模块是“存储原材料”，那决策模块就是Agent的“核心工厂”——它根据收集到的信息和记忆，自主判断“该怎么做”，是Agent智能的核心体现。

传统程序的“决策”是程序员写死的if-else，而Agent的决策模块是“动态的”——不用写死所有情况，它能根据具体场景自主拆分步骤、调整策略。

1. 决策模块要解决两个核心问题

决策模块的工作，本质上是解决两个问题：

• 做什么：把用户的目标拆成可执行的小步骤。比如用户目标是“分析销售数据并生成报告”，决策模块会拆成“1. 读取销售数据文件；2. 清洗数据（处理缺失值）；3. 统计销售额、订单量；4. 生成可视化图表；5. 撰写分析报告”；
• 怎么做：为每个步骤选择合适的执行方式。比如“读取销售数据文件”，决策模块会判断“文件是CSV格式，用pandas的read_csv函数”；如果文件是Excel格式，就决策“用pandas的read_excel函数”。

2. 决策模块的工作流程：目标解析→步骤拆分→策略选择→动态调整

决策模块的工作流程就像一个“项目经理”，接到一个大项目后，拆解任务、分配资源、监控进度、调整方案：

1. 目标解析：先理解用户的核心目标，比如用户说“我的快递没到”，核心目标是“查询快递状态并解决未送达问题”；
2. 步骤拆分：把核心目标拆成可执行的小步骤，比如“查询订单号→查询物流轨迹→判断物流状态→生成解决方案”；
3. 策略选择：为每个步骤选择策略，比如“如果物流显示滞留，策略是提交催单；如果物流显示已签收，策略是查询签收记录并联系快递员”；
4. 动态调整：根据执行结果调整策略，比如提交催单后，物流还是没动，决策模块会调整策略“联系人工客服跟进”。

3. 程序员视角：怎么实现一个简单的决策模块？

决策模块的实现难度相对高一些，因为需要结合大模型的语义理解能力。咱们用“大模型+规则”的混合方式，实现一个简单的决策模块：

from langchain.chat_models import ChatOpenAI

# 决策模块：结合大模型和规则，自主拆分步骤和选择策略
class DecisionModule:
    def __init__(self, llm_api_key):
        # 初始化大模型，用于语义理解和步骤拆分
        self.llm = ChatOpenAI(model_name="gpt-3.5-turbo", api_key=llm_api_key)
    
    # 核心方法：根据目标和上下文，生成决策（步骤+策略）
    def make_decision(self, goal, context, memory):
        # 1. 构建提示词：告诉大模型要做什么（拆分步骤、选择策略）
        prompt = f"""
        你是一个智能决策助手，需要帮Agent完成目标。请根据以下信息，做两件事：
        1. 把目标拆成3-5个可执行的步骤（按顺序排列）；
        2. 为每个步骤选择合适的执行策略（比如调用什么工具、执行什么操作）。
        
        目标：{goal}
        上下文信息：{context}
        相关记忆：{memory}
        
        输出格式：
        步骤1：[步骤描述]
        策略1：[执行策略]
        步骤2：[步骤描述]
        策略2：[执行策略]
        ...
        """
        # 2. 调用大模型生成决策
        decision_result = self.llm.predict(prompt)
        # 3. 解析大模型的输出，变成结构化数据（方便后续执行）
        structured_decision = self._parse_decision(decision_result)
        return structured_decision
    
    # 解析大模型输出，变成字典格式
    def _parse_decision(self, decision_text):
        steps = []
        strategies = []
        lines = decision_text.split("\n")
        for line in lines:
            line = line.strip()
            if line.startswith("步骤"):
                steps.append(line.split("：")[1].strip())
            elif line.startswith("策略"):
                strategies.append(line.split("：")[1].strip())
        # 组合成结构化数据
        return [{"步骤": step, "策略": strategy} for step, strategy in zip(steps, strategies)]

# 使用示例
if __name__ == "__main__":
    # 初始化决策模块（替换成你的API密钥）
    decision_module = DecisionModule("你的大模型API密钥")
    # 目标：用户需求
    goal = "帮我分析2024年3月的销售数据，生成可视化图表并写一份分析报告"
    # 上下文信息：感知模块处理后的信息（比如数据文件路径、数据格式）
    context = "数据文件路径：./sales_202403.csv，数据格式：CSV，包含字段：日期、产品类别、销售额、订单量"
    # 相关记忆：从记忆模块检索到的信息（比如之前的分析规则）
    memory = "记忆：销售数据分析需要统计Top5产品、月度销售额趋势、各品类占比"
    # 生成决策
    decision = decision_module.make_decision(goal, context, memory)
    print("生成的决策：")
    for item in decision:
        print(f"步骤：{item['步骤']}，策略：{item['策略']}")

这个决策模块的核心逻辑是：用大模型的语义理解能力拆分步骤和选择策略，再把输出解析成结构化数据。实际开发中，你可以加入“规则校验”，比如限制步骤数量、指定必须调用的工具，避免大模型生成不合理的决策；还可以加入“反馈机制”，如果某个步骤执行失败，让决策模块重新调整策略。

四、工具调用模块：Agent的“手脚”——执行决策，和外界交互

决策模块生成了“步骤和策略”，但Agent不能只“想”不“做”——工具调用模块就是Agent的“手脚”，核心作用是执行决策模块的策略，和外界交互，比如调用工具、操作文件、访问API、和用户沟通。

没有工具调用模块，Agent就是一个“只会想不会做”的“空想家”；有了它，Agent才能把决策落地，真正解决问题。

1. 工具调用模块要调用哪些“工具”？

Agent能调用的工具非常多，本质上是“所有能帮它完成任务的程序、接口、服务”，常见的有：

• 数据操作工具：数据库（MySQL、PostgreSQL）、文件系统（读/写文件）、Excel/CSV处理工具（pandas）；
• API接口工具：第三方API（天气API、物流API）、内部系统API（用户系统API、订单系统API）；
• 开发工具：代码编译器、静态代码分析工具（SonarQube）、测试工具（pytest）；
• 沟通工具：邮件服务、短信服务、聊天接口（Slack、钉钉）；
• 自定义工具：你自己写的Python/Java函数，比如“数据清洗函数”“加密函数”。

2. 工具调用模块的工作流程：工具选择→参数解析→执行→结果返回

工具调用模块的工作流程就像咱们调用函数：

1. 工具选择：根据决策模块的策略，选择对应的工具。比如策略是“读取CSV文件”，就选择“pandas的read_csv函数”；策略是“查询物流信息”，就选择“物流API”；
2. 参数解析：解析执行工具需要的参数。比如调用read_csv需要“文件路径”“编码格式”；调用物流API需要“订单号”“用户ID”；
3. 执行调用：调用工具并执行，比如执行read_csv函数读取文件，发送HTTP请求调用物流API；
4. 结果处理：接收工具的执行结果，比如读取到的数据、API返回的物流信息，然后把结果返回给决策模块和记忆模块（存储执行结果）。

3. 程序员视角：怎么实现一个简单的工具调用模块？

咱们用Python实现一个支持“读取CSV文件”“生成图表”“写报告”的工具调用模块：

import pandas as pd
import matplotlib.pyplot as plt
from typing import Any

# 第一步：定义各种工具（函数形式）
class Tools:
    # 工具1：读取CSV文件
    @staticmethod
    def read_csv(file_path: str, encoding: str = "utf-8") -> pd.DataFrame:
        try:
            df = pd.read_csv(file_path, encoding=encoding)
            return {"status": "success", "data": df, "message": "文件读取成功"}
        except Exception as e:
            return {"status": "fail", "data": None, "message": f"读取失败：{str(e)}"}
    
    # 工具2：生成可视化图表
    @staticmethod
    def generate_chart(df: pd.DataFrame, chart_type: str, save_path: str) -> str:
        try:
            plt.rcParams['font.sans-serif'] = ['SimHei']  # 解决中文显示问题
            if chart_type == "sales_trend":
                # 销售额趋势图
                df.plot(x="日期", y="销售额", kind="line")
                plt.title("月度销售额趋势")
            elif chart_type == "category_ratio":
                # 品类占比饼图
                category_sales = df.groupby("产品类别")["销售额"].sum()
                category_sales.plot(kind="pie", autopct="%1.1f%%")
                plt.title("各品类销售额占比")
            plt.savefig(save_path, dpi=300, bbox_inches="tight")
            plt.close()
            return {"status": "success", "data": save_path, "message": "图表生成成功"}
        except Exception as e:
            return {"status": "fail", "data": None, "message": f"图表生成失败：{str(e)}"}
    
    # 工具3：生成分析报告
    @staticmethod
    def generate_report(df: pd.DataFrame, report_path: str) -> str:
        try:
            # 计算核心指标
            total_sales = df["销售额"].sum()
            total_orders = df["订单量"].sum()
            top5_products = df.groupby("产品类别")["销售额"].sum().nlargest(5)
            # 撰写报告
            report = f"""
            2024年3月销售数据分析报告
            1. 整体业绩：
               - 总销售额：{total_sales:.2f}元
               - 总订单量：{total_orders}单
            2. Top5产品销售额：
               {top5_products.to_string()}
            3. 趋势分析：
               - 销售额整体呈上升趋势（具体需结合图表）
               - 家电品类占比最高（具体需结合图表）
            """
            with open(report_path, "w", encoding="utf-8") as f:
                f.write(report)
            return {"status": "success", "data": report_path, "message": "报告生成成功"}
        except Exception as e:
            return {"status": "fail", "data": None, "message": f"报告生成失败：{str(e)}"}

# 第二步：工具调用模块（管理工具、执行调用）
class ToolCallingModule:
    def __init__(self):
        self.tools = Tools()  # 加载所有工具
    
    # 核心方法：根据决策策略，调用对应的工具
    def call_tool(self, strategy: str, params: dict) -> dict:
        # 根据策略选择工具
        if "读取CSV文件" in strategy:
            # 调用读取CSV工具
            return self.tools.read_csv(
                file_path=params.get("file_path"),
                encoding=params.get("encoding", "utf-8")
            )
        elif "生成可视化图表" in strategy:
            # 调用生成图表工具
            return self.tools.generate_chart(
                df=params.get("df"),
                chart_type=params.get("chart_type"),
                save_path=params.get("save_path")
            )
        elif "生成分析报告" in strategy:
            # 调用生成报告工具
            return self.tools.generate_report(
                df=params.get("df"),
                report_path=params.get("report_path")
            )
        else:
            return {"status": "fail", "message": f"没有找到对应的工具：{strategy}"}

# 使用示例
if __name__ == "__main__":
    # 初始化工具调用模块
    tool_module = ToolCallingModule()
    # 1. 调用读取CSV工具
    read_result = tool_module.call_tool(
        strategy="读取CSV文件",
        params={"file_path": "./sales_202403.csv"}
    )
    if read_result["status"] == "success":
        df = read_result["data"]
        print("CSV文件读取成功")
        # 2. 调用生成图表工具
        chart_result = tool_module.call_tool(
            strategy="生成可视化图表",
            params={"df": df, "chart_type": "sales_trend", "save_path": "./sales_trend.png"}
        )
        print(chart_result["message"])
        # 3. 调用生成报告工具
        report_result = tool_module.call_tool(
            strategy="生成分析报告",
            params={"df": df, "report_path": "./sales_report.txt"}
        )
        print(report_result["message"])

这个工具调用模块的核心是“策略→工具的映射”，根据决策模块的策略，找到对应的工具并传入参数执行。实际开发中，你可以用LangChain的Tool类来封装工具，支持自动参数解析和类型校验；还可以加入“工具优先级”，比如同一个任务有多个工具可用时，选择效率最高的工具。

五、总结：Agent的核心模块是“流水线”，环环相扣

聊完四个核心模块，咱们可以用一张“流水线”图来理解它们的关系：
用户目标 → 感知模块（收集+预处理信息） → 记忆模块（存储+检索相关记忆） → 决策模块（拆分步骤+选择策略） → 工具调用模块（执行策略+返回结果） → 结果反馈给用户和记忆模块 → 循环直到达成目标

每个模块都有自己的核心作用，缺一不可：

• 感知模块：保证“信息全”；
• 记忆模块：保证“不健忘”；
• 决策模块：保证“方向对”；
• 工具调用模块：保证“能落地”。

对咱们程序员来说，开发Agent的本质就是“搭建这四个模块，并让它们高效协作”——你可以用现成的框架（比如LangChain）来快速集成这些模块，也可以根据业务需求自定义每个模块的实现。

下一节，咱们会深入每个模块的“进阶技巧”，比如怎么优化感知模块的信息采集效率、怎么让记忆模块支持语义检索、怎么让决策模块更智能，为后面的实战开发做更充分的准备。

目前国内还是很缺AI人才的，希望更多人能真正加入到AI行业，共同促进行业进步。想要系统学习AI知识的朋友可以看看我的教程http://blog.csdn.net/jiangjunshow，从深度学习基础原理到各领域实战应用都有讲解。