如何从0到1开发一个AI Agent

m0_62937412

341人浏览 · 2026-07-27 15:37:09

m0_62937412 · 2026-07-27 15:37:09 发布

1. 引言

随着大语言模型（LLM）能力的飞速提升，AI Agent（智能体）已成为人工智能领域最热门的方向之一。AI Agent 能够自主感知环境、制定计划、调用工具并执行任务，从简单的问答助手进化为真正的"数字员工"。本文将从零开始，带你一步步构建一个可运行的 AI Agent，涵盖核心概念、技术选型、代码实现和最佳实践。

2. 理解 AI Agent 的核心架构

在动手编码之前，我们需要先理解 AI Agent 的基本组成。一个典型的 AI Agent 包含以下核心模块：

感知模块：接收用户输入或环境信息，理解当前状态。
推理与规划模块：基于 LLM 分析任务，拆解为可执行的子步骤。
记忆模块：短期记忆（对话上下文）和长期记忆（向量数据库存储的知识）。
工具调用模块：执行具体操作，如搜索网页、调用 API、读写文件等。
反馈与迭代模块：根据执行结果调整策略，直到完成任务。

这些模块协同工作，使 Agent 能够像人类一样"思考-行动-观察-调整"。

3. 技术选型与环境准备

我们将使用以下技术栈来构建 AI Agent：

编程语言：Python 3.10+
LLM 接口：OpenAI API（兼容接口，如 GPT-4o）
Agent 框架：LangChain（社区成熟，生态丰富）
向量数据库：ChromaDB（轻量级，适合本地开发）
工具库：Requests、BeautifulSoup（网页抓取）、Tavily Search API（搜索）

首先安装必要的依赖：

pip install langchain langchain-openai chromadb tavily-python python-dotenv

创建 .env 文件配置 API 密钥：

OPENAI_API_KEY=your_openai_api_key
TAVILY_API_KEY=your_tavily_api_key

4. 构建基础 Agent：从简单对话开始

我们先实现一个最简版本的 Agent，它能够理解用户意图并调用一个工具。以下代码展示了 Agent 的核心循环：

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain.prompts import PromptTemplate
load_dotenv()
1. 初始化 LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)
2. 定义工具
@tool
def get_current_time() -> str:
"""获取当前系统时间"""
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
3. 创建 Agent
tools = [get_current_time]
prompt = PromptTemplate.from_template(
"你是一个智能助手。请使用以下工具来回答用户的问题。\n"
"工具列表：{tools}\n"
"工具名称：{tool_names}\n"
"用户输入：{input}\n"
"思考过程：{agent_scratchpad}"
)
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
4. 运行 Agent
response = agent_executor.invoke({"input": "现在几点了？"})
print(response["output"])

这段代码演示了 Agent 的四个关键步骤：初始化 LLM、定义工具、创建 Agent 执行器、运行推理。运行后，Agent 会自动判断需要调用 get_current_time 工具，并将结果返回给用户。

5. 增强 Agent：添加搜索与记忆能力

基础 Agent 只能调用本地工具，接下来我们为其添加网络搜索和对话记忆能力，使其更像一个真正的智能助手。

from langchain.memory import ConversationBufferMemory
from langchain.agents import Tool
from langchain_community.tools.tavily_search import TavilySearchResults
搜索工具
search_tool = TavilySearchResults(max_results=3)
包装为 LangChain Tool
tools = [
Tool(
name="web_search",
func=search_tool.run,
description="搜索互联网获取最新信息，输入为搜索关键词"
),
Tool(
name="get_time",
func=get_current_time,
description="获取当前系统时间"
)
]
添加对话记忆
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
创建带记忆的 Agent
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True,
max_iterations=5  # 防止无限循环
)
多轮对话测试
agent_executor.invoke({"input": "帮我搜索一下2025年诺贝尔物理学奖得主"})
agent_executor.invoke({"input": "刚才提到的获奖者主要贡献是什么？"})

通过 ConversationBufferMemory，Agent 能够记住前一轮对话的上下文，实现连贯的多轮交互。max_iterations 参数则防止 Agent 陷入死循环。

6. 进阶：实现自主规划与任务分解

真正的 AI Agent 需要具备复杂任务分解能力。下面我们实现一个"研究助手"Agent，它能将用户的问题拆解为多个子任务并依次执行：

from langchain.chains import LLMChain
from langchain.prompts import ChatPromptTemplate
任务分解提示词
planner_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个任务规划专家。将用户的问题拆解为最多3个可执行的子任务，每个子任务应包含明确的搜索关键词。"),
("human", "用户问题：{input}\n请输出JSON格式的任务列表，格式：[{{'task': '描述', 'query': '搜索关键词'}}]")
])
planner_chain = LLMChain(llm=llm, prompt=planner_prompt)
执行规划
plan = planner_chain.run(input="分析2025年AI行业的主要趋势")
print("任务规划：", plan)
依次执行子任务
import json
tasks = json.loads(plan)
results = []
for task in tasks:
search_result = search_tool.run(task["query"])
results.append({"task": task["task"], "result": search_result})
汇总结果
summary_prompt = ChatPromptTemplate.from_messages([
("system", "你是一个分析报告撰写专家。基于以下子任务结果，生成一份完整的分析报告。"),
("human", "子任务结果：{results}")
])
summary_chain = LLMChain(llm=llm, prompt=summary_prompt)
final_report = summary_chain.run(results=json.dumps(results, ensure_ascii=False))
print(final_report)

这种"规划-执行-汇总"模式是高级 Agent 的典型架构。Agent 先利用 LLM 的推理能力制定计划，然后按计划调用工具，最后整合结果输出。

7. 部署与生产化建议

将 Agent 部署到生产环境时，需要考虑以下关键点：

错误处理：添加重试机制和降级策略，当 API 调用失败时自动切换备用方案。
速率限制：使用令牌桶算法控制 API 调用频率，避免触发限流。
安全审计：对工具调用进行白名单控制，防止 Agent 执行危险操作。
监控与日志：记录每次 Agent 的思考过程和工具调用结果，便于调试和优化。
成本控制：设置每次对话的 Token 上限，使用缓存减少重复调用。

推荐使用 FastAPI 将 Agent 包装为 RESTful 服务：

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Query(BaseModel):
text: str
session_id: str = "default"
@app.post("/agent/chat")
async def chat(query: Query):
# 根据 session_id 管理不同会话的记忆
response = agent_executor.invoke({"input": query.text})
return {"response": response["output"]}