AI Agent 开发实战完全指南：从零构建智能任务执行系统

AI Agent 是 2026 年最热门的技术方向，它让 AI 从被动回答问题进化为主动执行任务。本文将带你从零开始构建一个完整的 AI Agent，掌握工具调用、记忆管理、多 Agent 协作等核心技能。

一、什么是 AI Agent？为什么它如此重要？

传统的 AI 应用就像一个只会回答问题的百科全书，你问什么，它答什么。而 AI Agent 则更像一个会干活的助理，你只需要告诉它目标，它会自己规划步骤、调用工具、执行任务，直到完成目标。

AI Agent 的核心特征：

• 自主决策：能够理解复杂目标，拆解成可执行的子任务
• 工具调用：可以调用 API、执行代码、操作数据库、浏览网页
• 记忆管理：记住上下文，保持任务的连续性和可追溯性
• 动态调整：根据执行结果反馈，实时调整策略

举个实际的例子：你告诉 Agent「帮我分析竞品的价格策略，并生成一份报告」。

传统 AI 只能基于已有知识给你一些建议。但 Agent 会这样工作：

1. 调用搜索 API，获取竞品官网
2. 爬取价格页面，提取关键信息
3. 调用数据分析工具，生成对比图表
4. 使用文档生成工具，输出完整报告

这就是 Agent 的魅力——把 AI 从「聊天机器人」升级为「任务执行者」。

二、AI Agent 的核心架构

一个完整的 AI Agent 系统包含四个核心模块：

1. LLM（大语言模型）
Agent 的「大脑」，负责理解用户意图、规划任务、决策下一步行动。GPT-4、Claude、文心一言等都是常用的 LLM。

2. Tools（工具集）
Agent 的「手脚」，让 Agent 能与外部世界交互。常见工具包括：

• 搜索工具：调用搜索引擎获取信息
• 代码执行器：运行 Python 代码进行计算
• 数据库工具：查询和操作数据库
• 文件工具：读写文件、生成文档
• API 工具：调用第三方服务

3. Memory（记忆系统）
Agent 的「记忆力」，存储对话历史、任务状态、执行结果。分为短期记忆（当前会话）和长期记忆（持久化存储）。

4. Planner（规划器）
Agent 的「策略中心」，将复杂目标拆解为执行计划，并根据反馈动态调整。

三、从零构建一个 AI Agent

接下来，我们用 Python 从零构建一个简单但完整的 AI Agent。

3.1 环境准备

# 安装依赖
# pip install openai python-dotenv

import os
import json
from openai import OpenAI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 初始化 OpenAI 客户端
client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)

3.2 定义工具

我们定义几个实用的工具：搜索、计算、获取天气。

# 定义工具函数
def search_web(query: str) -> str:
    """
    搜索网络信息
    :param query: 搜索关键词
    :return: 搜索结果摘要
    """
    # 这里简化实现，实际可调用 Google/Bing 搜索 API
    results = {
        "Python 教程": "Python 是最流行的编程语言，适合初学者入门...",
        "AI Agent": "AI Agent 是能够自主执行任务的智能体...",
        "RAG 技术": "RAG 是检索增强生成技术，结合知识库和 LLM..."
    }
    
    for key, value in results.items():
        if key in query:
            return value
    
    return f"未找到关于 '{query}' 的相关信息，建议使用更精确的关键词。"

def calculate(expression: str) -> str:
    """
    执行数学计算
    :param expression: 数学表达式
    :return: 计算结果
    """
    try:
        # 注意：生产环境应使用更安全的计算方式
        result = eval(expression)
        return f"计算结果: {expression} = {result}"
    except Exception as e:
        return f"计算错误: {str(e)}"

def get_weather(city: str) -> str:
    """
    获取城市天气
    :param city: 城市名称
    :return: 天气信息
    """
    # 模拟天气数据，实际可调用天气 API
    weather_data = {
        "北京": "晴天，温度 18°C，空气质量良好",
        "上海": "多云，温度 22°C，有轻微雾霾",
        "深圳": "小雨，温度 26°C，湿度较高"
    }
    
    return weather_data.get(city, f"未找到 {city} 的天气信息")

3.3 工具注册与描述

# 将工具注册为 OpenAI Function Calling 格式
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "搜索网络获取信息，适用于需要最新数据或特定知识的情况",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词，如 'Python 教程' 或 'AI Agent 开发'"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "执行数学计算，支持加减乘除和复杂表达式",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '(100 + 50) * 2' 或 'sqrt(16)'"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的实时天气信息",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如 '北京'、'上海'"
                    }
                },
                "required": ["city"]
            }
        }
    }
]

# 工具映射表（用于执行）
tool_functions = {
    "search_web": search_web,
    "calculate": calculate,
    "get_weather": get_weather
}

3.4 Agent 主循环

class AIAgent:
    def __init__(self, model="gpt-4o-mini"):
        """
        初始化 Agent
        :param model: 使用的模型名称
        """
        self.model = model
        self.messages = []  # 记忆系统：存储对话历史
        self.max_iterations = 5  # 最大迭代次数，防止无限循环
    
    def add_message(self, role, content):
        """
        添加消息到记忆
        :param role: 角色（user/assistant/system）
        :param content: 消息内容
        """
        self.messages.append({"role": role, "content": content})
    
    def run(self, user_input: str) -> str:
        """
        运行 Agent，处理用户输入
        :param user_input: 用户输入
        :return: 最终回复
        """
        # 添加用户消息
        self.add_message("user", user_input)
        
        # Agent 主循环
        for iteration in range(self.max_iterations):
            print(f"\n=== 第 {iteration + 1} 轮对话 ===")
            
            # 调用 LLM
            response = client.chat.completions.create(
                model=self.model,
                messages=self.messages,
                tools=tools,
                tool_choice="auto"  # 让模型决定是否调用工具
            )
            
            assistant_message = response.choices[0].message
            
            # 检查是否需要调用工具
            if assistant_message.tool_calls:
                # 添加助手消息到历史
                self.messages.append(assistant_message)
                
                # 执行所有工具调用
                for tool_call in assistant_message.tool_calls:
                    function_name = tool_call.function.name
                    function_args = json.loads(tool_call.function.arguments)
                    
                    print(f"调用工具: {function_name}")
                    print(f"参数: {function_args}")
                    
                    # 执行工具
                    if function_name in tool_functions:
                        result = tool_functions[function_name](**function_args)
                    else:
                        result = f"错误: 未知工具 {function_name}"
                    
                    print(f"结果: {result}")
                    
                    # 将结果添加到消息历史
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": result
                    })
                
                # 继续下一轮，让模型处理工具结果
                continue
            
            # 如果没有工具调用，说明任务完成
            final_response = assistant_message.content
            self.add_message("assistant", final_response)
            return final_response
        
        # 达到最大迭代次数
        return "抱歉，我无法在限定步数内完成任务，请简化您的问题或分步骤提问。"

3.5 测试 Agent

# 创建 Agent 实例
agent = AIAgent(model="gpt-4o-mini")

# 测试 1：简单问题
print("\n测试 1: 简单问题")
response = agent.run("你好，请介绍一下你自己")
print(f"\n回复: {response}")

# 测试 2：需要调用工具的问题
print("\n\n测试 2: 调用工具")
agent2 = AIAgent(model="gpt-4o-mini")
response = agent2.run("帮我搜索 AI Agent 的相关信息，然后计算 100 乘以 50 是多少")
print(f"\n回复: {response}")

# 测试 3：多步骤任务
print("\n\n测试 3: 多步骤任务")
agent3 = AIAgent(model="gpt-4o-mini")
response = agent3.run("查一下北京的天气，然后计算 (18 + 22) / 2")
print(f"\n回复: {response}")

四、进阶：多 Agent 协作

单个 Agent 的能力有限，复杂任务需要多个 Agent 协作完成。这就是多智能体系统（Multi-Agent System）。

4.1 CrewAI 框架简介

CrewAI 是 2026 年最热门的多 Agent 框架，它通过角色分工和流程设计，让多个 Agent 协同工作。

# 安装 CrewAI
# pip install crewai

from crewai import Agent, Task, Crew, Process

# 定义研究 Agent
researcher = Agent(
    role="研究员",
    goal="收集和整理相关信息",
    backstory="你是一位经验丰富的研究员，擅长从海量信息中提取关键内容",
    verbose=True
)

# 定义写作 Agent
writer = Agent(
    role="技术作家",
    goal="将研究结果转化为清晰易懂的文章",
    backstory="你是一位资深技术作家，擅长将复杂概念讲解得通俗易懂",
    verbose=True
)

# 定义审核 Agent
reviewer = Agent(
    role="内容审核",
    goal="确保文章质量，检查错误和改进建议",
    backstory="你是一位严谨的编辑，对细节一丝不苟",
    verbose=True
)

# 定义任务
research_task = Task(
    description="研究 AI Agent 的最新发展趋势和应用案例",
    agent=researcher
)

write_task = Task(
    description="根据研究结果撰写一篇技术文章",
    agent=writer
)

review_task = Task(
    description="审核文章，提出改进建议",
    agent=reviewer
)

# 创建团队并执行
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    process=Process.sequential  # 顺序执行
)

# 运行
result = crew.kickoff()
print(result)

4.2 记忆系统的实现

生产环境的 Agent 需要持久化记忆，这里我们实现一个基于向量数据库的记忆系统。

import chromadb
from chromadb.utils import embedding_functions

class AgentMemory:
    def __init__(self, collection_name="agent_memory"):
        """
        初始化向量数据库记忆系统
        :param collection_name: 集合名称
        """
        # 初始化 Chroma 客户端
        self.client = chromadb.Client()
        
        # 使用 OpenAI embedding
        self.embedding_function = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.getenv("OPENAI_API_KEY"),
            model_name="text-embedding-3-small"
        )
        
        # 创建集合
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_function
        )
    
    def store(self, text: str, metadata: dict = None):
        """
        存储记忆
        :param text: 记忆内容
        :param metadata: 元数据（如时间戳、来源等）
        """
        import uuid
        self.collection.add(
            documents=[text],
            metadatas=[metadata or {}],
            ids=[str(uuid.uuid4())]
        )
        print(f"已存储记忆: {text[:50]}...")
    
    def recall(self, query: str, n_results: int = 5) -> list:
        """
        检索相关记忆
        :param query: 查询内容
        :param n_results: 返回结果数量
        :return: 相关记忆列表
        """
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )
        return results["documents"][0] if results["documents"] else []

# 使用示例
memory = AgentMemory()

# 存储记忆
memory.store("用户偏好使用 Python 进行数据分析", {"type": "preference"})
memory.store("上次任务：分析销售数据，生成了月度报告", {"type": "task_history"})

# 检索记忆
related = memory.recall("数据分析")
print(f"相关记忆: {related}")

五、生产环境最佳实践

5.1 安全性考虑

• 工具权限控制：限制 Agent 可调用的工具范围，避免敏感操作
• 输入验证：对用户输入和工具参数进行严格校验
• 执行沙箱：代码执行应在隔离环境中进行
• 审计日志：记录所有 Agent 行为，便于追溯

5.2 性能优化

• 模型选择：简单任务用小模型（GPT-4o-mini），复杂推理用大模型（GPT-4o）
• 缓存策略：对重复查询进行缓存，减少 API 调用
• 并行执行：多个独立工具调用可并行处理
• 流式输出：使用流式响应提升用户体验

5.3 成本控制

• 限制迭代次数：设置最大步数，防止无限循环消耗 token
• 精简 Prompt：优化系统提示词，减少不必要的 token 消耗
• 多模型策略：75% 的团队使用多模型策略，根据任务复杂度选择合适的模型

六、总结

AI Agent 正在改变我们与 AI 交互的方式。从被动的问答工具，进化为主动的任务执行者。掌握 Agent 开发，意味着你站在了 AI 应用开发的最前沿。

关键要点回顾：

1. 核心架构：LLM + Tools + Memory + Planner 是 Agent 的四大支柱
2. 工具调用：通过 Function Calling 让 Agent 具备执行能力
3. 记忆系统：向量数据库实现长期记忆，确保任务连续性
4. 多 Agent 协作：CrewAI 等框架让复杂任务分工协作
5. 生产实践：安全性、性能、成本是需要重点关注的三个方面

2026 年是 AI Agent 从概念验证走向企业级落地的关键一年。现在开始学习，你就是这场变革的先行者。

本文链接：https://www.kkkliao.cn/?id=946 转载需授权！