（源码+实例）大家都在说Agent，那么Agent到底是什么？

前言

在了解Agent之前，我们先了解一下LLM；

我们都知道22年11月30日，OpenAI发布了ChatGPT，而ChatGPT就是LLM的典型的应用实例之一，包括后面发布的文心一言、通义千问等大模型。

LLM全称Large Language Model，中文为大语言模型，是人工智能领域中自然语言处理(NLP)方向的核心技术，也是当前生成式AI的底层支撑。它的本质是基于海量文本数据训练、具备强大语言理解与生成能力的深度学习模型。

我们从以上的概念中很容易就能看到LLM的一些局限，比如：

• 语言理解：只能基于训练过的文本数据去进行理解
• 语言生成：只能以文本的方式去进行交互

这两个局限使得LLM更像是一个建议者、开车时的副驾驶，无法执行具体的动作。

而在AI的发展中，我们更需要AI去做一个执行者，去做司机，去实际参与物理世界，由此，Agent发展出来了。

概念

现在我们常常用这样的一个智能体架构公式去理解Agent:

有了 LLM(大语言模型)、Perception(感知)、Planning(规划)、Memory(记忆) 、Tools(工具)，Agent真正具备了五官与手脚，去理解、操作我们的物理世界：

• LLM(大语言模型）：Agent的大脑
  • 具体可参考前言部分
• Perception(感知)：Agent与外部沟通的桥梁，通过多种渠道感知环境信息，类似于人的感官
  • 视觉感知（CV）
    • 通过摄像头或图像输入，看到并理解内容
  • 听觉感知(ASR)
    • 通过麦克风或音频输入，听到并识别语音内容
  • 其它感知
    • 集成传感器，感知环境的物理状态，如温度、湿度、味道等
• Planning(规划)：Agent的决策执行引擎，负责将LLM拆解的目标转化为可执行的分布计划，并根据反馈动态调整策略
  • 任务拆解
    • 将复杂目标拆解成一系列可执行的简单子任务
  • 制定策略
    • 为每个子任务规划出具体的执行步骤和方法
  • 动态调整
    • 根据执行反馈灵活调整计划，应对突发情况
• Memory(记忆)：Agent的长期知识库，解决LLM上下文窗口有限的问题，为Agent提供长期学习和经验积累的能力
  • 短期记忆
    • 类似工作记忆，存储当前任务上下文，支持多轮对话与任务连贯性
  • 长期记忆
    • 类似知识库，持久化存储知识、偏好与经验，可随时检索利用
• Tools(工具)：Agent的双手，让Agent突破纯文本能力限制，实现"知行合一"的任务执行
  • 访问信息
    • 通过搜索引擎获取最新、最准确的信息
  • 执行计算
    • 使用计算器或代码编辑器完成复杂运算和数据分析
  • 控制设备
    • 通过API调用，控制智能家居、工业机器人等物理设备
  • 操作软件
    • 自动操作电脑软件，如Excel、Photoshop等

Demo

了解了上面的概念，我们来实操一下，搭建一个本地的Agent；

Demo介绍

功能

实现的Agent的功能点包括：

• 理解用户的自然语言问题
• 自动判断是否需要调用外部工具
• 执行 Python 函数（如获取时间、数学计算求和）
• 将工具执行结果整合成人类可读的回答

技术栈

基于ReAct结构，用通义千问大模型作为AI基座，Python3作为开发语言：

• 大模型：阿里云百炼（通义千问）
• 通信协议：OpenAI 兼容模式
• 编程语言：Python 3.8+

文件目录

Demo实现

第一步：获取API Key

我这里用的是阿里云的大模型：

• 访问 阿里云百炼控制台
• 注册/登录账号
• 创建 API Key

第二步：安装依赖

创建 requirements.txt 文件：

openai
python-dotenv

安装依赖：

pip3 install -r requirements.txt

第三步：配置环境变量

创建 .env 文件：

DASHSCOPE_API_KEY=你的API_KEY

第四步：实现工具转换器（utils.py）

创建utils.py文件，利用Python 的 inspect 模块，读取函数的函数名、参数列表和类型注解，生成大模型可理解的JSON Schema，让大模型去了解工具的功能和参数

import inspect

def function_to_json(func) -> dict:
    """
    通过 Python 反射机制，自动将一个 Python 函数转换为 OpenAI/DashScope 兼容的工具定义 (JSON Schema)。
    这使得我们只需要编写普通的 Python 函数，就能直接被 Agent 识别和调用。
    """
    # 获取函数的签名信息（包含参数名、参数类型注解、默认值等）
    sig = inspect.signature(func)
    parameters = {}
    required = []

    # 遍历函数的所有参数
    for name, param in sig.parameters.items():
        # 根据 Python 的类型注解推断 JSON Schema 的类型
        if param.annotation == float:
            param_type = "number"
        elif param.annotation == int:
            param_type = "integer"
        elif param.annotation == str:
            param_type = "string"
        elif param.annotation == bool:
            param_type = "boolean"
        else:
            # 默认设为 string
            param_type = "string"

        # 构造单个参数的 schema 定义
        parameters[name] = {"type": param_type}

        # 如果参数没有设定默认值，则在 Schema 中标记为必填项 (required)
        if param.default is inspect.Parameter.empty:
            required.append(name)

    # 返回符合 OpenAI Tool 定义规范的字典结构
    return {
        "type": "function",
        "function": {
            "name": func.__name__,                      # 函数名
            "description": inspect.getdoc(func) or "",  # 使用函数的 docstring 作为工具描述
            "parameters": {
                "type": "object",
                "properties": parameters,                # 参数列表
                "required": required,                    # 必填参数列表
            },
        },
    }

第五步：定义工具函数（tools/）

关键点：

• 每个函数必须有清晰的 docstring（模型用它来理解工具用途）
• 使用 类型注解标注参数类型
• 函数名要有语义，便于模型选择

5.1 时间工具（tools/time_tool.py）

from datetime import datetime


def get_current_datetime() -> str:
    """
    获取当前日期和时间的工具。
    Agent 可以调用此工具来了解当前的实时时间。
    :return: 格式化后的日期时间字符串，例如 "2024-01-05 12:00:00"。
    """
    current_datetime = datetime.now()
    formatted_datetime = current_datetime.strftime("%Y-%m-%d %H:%M:%S")
    return formatted_datetime

5.2 数学工具（tools/math_tools.py）

def add(a: int, b: int) -> int:
    """
    计算两个整数的和。
    Agent 会在需要进行加法运算时自动调用此工具。
    """
    return a + b

第六步：实现 Agent 核心逻辑（agent.py）

• 维护对话历史（Memory - 短期记忆）
• 调用大模型（LLM）
• 解析工具调用指令（Planning - 任务规划）
• 执行工具并反馈结果（Tools - 工具执行）

6.1 初始化

初始化系统提示词，创建工具映射

import json
from typing import List, Dict, Any
from openai import OpenAI
from utils import function_to_json

SYSTEM_PROMPT = """
你是一个人工智能助手。你的输出应该与用户的语言保持一致。
当用户的问题需要调用工具时，你可以从提供的工具列表中调用适当的工具函数。
"""

class Agent:
    def __init__(self, client: OpenAI, model: str = "qwen-plus", 
                 tools: List = None, verbose: bool = True):
        self.client = client
        self.tools = tools or []
        self.model = model
        # Memory: 初始化对话历史，包含系统提示词
        self.messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        self.verbose = verbose
        # Tools: 创建工具映射：函数名 -> 函数对象
        self.tool_map = {tool.__name__: tool for tool in self.tools}

6.2 工具 Schema 生成

将时间工具和数学工具转换为模型可识别的JSON Schema

def get_tool_schema(self) -> List[Dict[str, Any]]:
    """将 Python 函数列表转换为模型可识别的 JSON Schema"""
    return [function_to_json(tool) for tool in self.tools]

6.3 工具调用处理

处理模型发出的单个工具调用逻辑

def handle_tool_call(self, tool_call):
    """处理模型发出的单个工具调用请求"""
    function_name = tool_call.function.name
    function_args_str = tool_call.function.arguments
    function_id = tool_call.id

    try:
        # 解析 JSON 字符串参数
        function_args = json.loads(function_args_str)
        
        # 从映射表中查找并执行函数
        if function_name in self.tool_map:
            if self.verbose:
                print(f">>> 正在执行工具: {function_name}, 参数: {function_args}")
            
            func = self.tool_map[function_name]
            result = func(**function_args)
        else:
            result = f"错误: 找不到名为 {function_name} 的工具"
    except Exception as e:
        result = f"工具执行过程中出错: {str(e)}"

    return {
        "role": "tool",
        "content": str(result),
        "tool_call_id": function_id,
    }

6.4 ReAct 循环（核心逻辑 - Planning）

处理用户输入的核心方法，实现 ReAct 循环：

实现 Planning（规划）的核心思想：

• 任务拆解：将用户问题拆解为可执行的工具调用
• 制定策略：决定调用哪些工具、以什么顺序调用
• 动态调整：根据工具执行结果调整后续策略

def get_completion(self, prompt: str) -> str:
   
    # Memory: 添加用户问题到历史
    self.messages.append({"role": "user", "content": prompt})

    # Planning: 第一次请求 - LLM 判断是否需要工具
    response = self.client.chat.completions.create(
        model=self.model,
        messages=self.messages,
        tools=self.get_tool_schema() if self.tools else None,
        stream=False,
    )

    message = response.choices[0].message

    # Planning: 循环处理工具调用（支持多次调用或链式调用）
    while message.tool_calls:
        # Memory: 将模型的工具调用请求加入历史
        self.messages.append(message)

        # Tools: 执行所有请求的工具
        for tool_call in message.tool_calls:
            tool_result_message = self.handle_tool_call(tool_call)
            # Memory: 将工具执行结果加入历史
            self.messages.append(tool_result_message)

        # Planning: 带着工具结果再次请求模型，进行动态调整
        response = self.client.chat.completions.create(
            model=self.model,
            messages=self.messages,
            tools=self.get_tool_schema() if self.tools else None,
            stream=False,
        )
        message = response.choices[0].message

    # Memory: 模型不再调用工具，返回最终回复，存入历史并返回
    self.messages.append({"role": "assistant", "content": message.content})
    return message.content or ""

第七步：创建主程序（main.py）

• 初始化OpenAI
• 创建Agent实例，并注入工具
• 交互式对话循环

import os
from dotenv import load_dotenv
from openai import OpenAI
from agent import Agent
from tools.time_tool import get_current_datetime
from tools.math_tools import add, compare, count_letter_in_string

load_dotenv()

API_KEY = os.getenv("DASHSCOPE_API_KEY")

if __name__ == "__main__":
    if not API_KEY:
        print("错误: 请在 .env 文件中配置 DASHSCOPE_API_KEY")
        exit(1)

    # 初始化 OpenAI 客户端（使用阿里云百炼的兼容接口）
    client = OpenAI(
        api_key=API_KEY,
        base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    )

    # 创建 Agent 实例，注入工具
    agent = Agent(
        client=client,
        model="qwen-plus",
        tools=[get_current_datetime, add, compare, count_letter_in_string],
        verbose=True,
    )

    print("JUN 人工智能助手已启动！(输入 'exit' 退出)")

    # 交互式对话循环
    while True:
        try:
            prompt = input("33[94mUser: 33[0m")
            
            if prompt.lower() in ["exit", "quit", "退出"]:
                print("再见！")
                break

            if not prompt.strip():
                continue

            response = agent.get_completion(prompt)
            print("33[92mAssistant: 33[0m", response, "n")

        except KeyboardInterrupt:
            print("n程序终止")
            break
        except Exception as e:
            print(f"33[91m发生错误: {e}33[0m")

总结

以上是Agent的全部内容了，我们了解了LLM、Agent的基本概念，还简单在本地搭建了一个简易的Agent，用于帮你充分理解Agent的五大组成部分：LLM、感知、工具、记忆、规划，希望能对你有所帮助。