1. 什么是 AI Agent

AI Agent（智能体）是一种能够感知环境、做出决策并执行动作的智能系统。与传统的大语言模型（LLM）不同，Agent 不仅仅是生成文本，它能够调用工具、访问外部数据、规划任务步骤，并在多轮交互中自主完成复杂目标。

一个典型的 AI Agent 通常包含以下核心组件：

• 大语言模型（LLM）：作为推理和决策的“大脑”，负责理解用户意图、生成计划和回复。
• 工具调用（Tool Calling）：Agent 通过调用外部 API、数据库、搜索引擎或代码解释器来获取信息和执行操作。
• 记忆（Memory）：包括短期记忆（对话上下文）和长期记忆（向量数据库、知识库），用于跨会话保持状态。
• 规划（Planning）：将复杂任务拆解为多个子步骤，并决定执行顺序。
• 执行与反馈（Execution & Feedback）：执行动作后观察结果，并根据反馈调整下一步策略。

简单来说，LLM 是“会说话的模型”，而 Agent 是“会做事的系统”。

2. 环境准备

在开始编写 Agent 之前，需要准备好开发环境。推荐使用 Python 3.10 及以上版本，并安装必要的依赖库。

# 创建虚拟环境
python -m venv agent_env
source agent_env/bin/activate  # Windows 下使用 agent_env\Scripts\activate
安装核心依赖
pip install openai langchain langchain-openai python-dotenv
验证安装
python -c "import openai; print('OpenAI SDK 版本:', openai.version)"

接下来，在项目根目录创建 .env 文件，存放 API 密钥：

OPENAI_API_KEY=sk-your-api-key-here
OPENAI_BASE_URL=https://api.openai.com/v1

使用 python-dotenv 加载环境变量：

from dotenv import load_dotenv
import os
load_dotenv()
api_key = os.getenv("OPENAI_API_KEY")
base_url = os.getenv("OPENAI_BASE_URL")
print("API Key 已加载:", bool(api_key))

3. 第一个 Agent：调用工具

工具调用（Function Calling）是 Agent 的核心能力。下面我们实现一个简单的天气查询 Agent，它能够根据用户输入的城市名称调用天气 API。

首先，定义一个模拟的天气查询函数：

import json
import random
def get_weather(city: str) -> str:
"""模拟查询城市天气"""
weather_data = {
"北京": {"温度": 25, "天气": "晴"},
"上海": {"温度": 28, "天气": "多云"},
"广州": {"温度": 30, "天气": "小雨"},
}
if city in weather_data:
data = weather_data[city]
return json.dumps({"城市": city, **data}, ensure_ascii=False)
return json.dumps({"错误": f"未找到城市 {city} 的天气数据"}, ensure_ascii=False)

然后，定义工具描述并让模型决定何时调用：

from openai import OpenAI
client = OpenAI(api_key=api_key, base_url=base_url)
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "查询指定城市的实时天气",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，如北京、上海"}
},
"required": ["city"]
}
}
}
]
def run_agent(user_input: str):
messages = [{"role": "user", "content": user_input}]
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)
assistant_message = response.choices[0].message
如果模型决定调用工具
if assistant_message.tool_calls:
for tool_call in assistant_message.tool_calls:
if tool_call.function.name == "get_weather":
args = json.loads(tool_call.function.arguments)
result = get_weather(args["city"])
messages.append(assistant_message)
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result
})
# 将工具结果返回给模型生成最终回复
final_response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages
)
return final_response.choices[0].message.content
return assistant_message.content
测试
print(run_agent("北京今天天气怎么样？"))

运行上述代码，Agent 会自动识别需要调用 get_weather 工具，获取天气数据后生成自然语言回复。

4. 使用 LangChain 构建 Agent

LangChain 提供了更高层的抽象，让 Agent 开发更加便捷。下面使用 LangChain 的 create_tool_calling_agent 构建一个支持多工具的 Agent。

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
初始化模型
llm = ChatOpenAI(
model="gpt-4o",
api_key=api_key,
base_url=base_url,
temperature=0
)
定义工具
@tool
def add(a: float, b: float) -> float:
"""计算两个数字的和"""
return a + b
@tool
def multiply(a: float, b: float) -> float:
"""计算两个数字的乘积"""
return a * b
@tool
def get_weather(city: str) -> str:
"""查询城市天气"""
weather_map = {"北京": "晴 25°C", "上海": "多云 28°C"}
return weather_map.get(city, "暂无数据")
创建提示模板
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个乐于助人的 AI 助手，可以调用工具来完成任务。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
创建 Agent
tools = [add, multiply, get_weather]
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
运行 Agent
result = agent_executor.invoke({"input": "北京天气怎么样？顺便算一下 23 乘以 4 等于多少？"})
print(result["output"])

LangChain 自动处理了工具调用的循环逻辑，包括多轮工具调用、错误处理和结果汇总，大大简化了开发流程。

5. 为 Agent 添加记忆

记忆能力让 Agent 能够记住对话历史，实现多轮交互。LangChain 提供了多种记忆实现，下面演示使用 ConversationBufferMemory 和向量数据库实现短期与长期记忆。

5.1 短期记忆：对话缓冲

from langchain.memory import ConversationBufferMemory
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
memory = ConversationBufferMemory(
memory_key="chat_history",
return_messages=True
)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个有帮助的助手。"),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
agent=agent,
tools=tools,
memory=memory,
verbose=True
)
第一轮对话
agent_executor.invoke({"input": "我叫小明，请记住我的名字。"})
第二轮对话，Agent 能记住用户名字
result = agent_executor.invoke({"input": "我叫什么名字？"})
print(result["output"])

5.2 长期记忆：向量数据库

对于跨会话的长期记忆，可以使用向量数据库存储和检索信息。下面使用 Chroma 作为向量存储：

pip install chromadb langchain-community

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import CharacterTextSplitter
初始化嵌入模型
embeddings = OpenAIEmbeddings(
api_key=api_key,
base_url=base_url,
model="text-embedding-3-small"
)
创建向量数据库
vectorstore = Chroma(
collection_name="agent_memory",
embedding_function=embeddings,
persist_directory="./chroma_db"
)
存储记忆
def save_memory(content: str):
"""将重要信息存入长期记忆"""
vectorstore.add_texts([content])
vectorstore.persist()
def retrieve_memory(query: str, k: int = 3) -> list:
"""检索相关记忆"""
results = vectorstore.similarity_search(query, k=k)
return [doc.page_content for doc in results]
示例：保存用户偏好
save_memory("用户小明喜欢简洁的回答风格，偏好 Python 技术栈。")
save_memory("用户小明的项目是电商平台，使用 Django 框架。")
检索记忆
memories = retrieve_memory("用户喜欢什么技术栈？")
print("检索到的记忆:", memories)

6. Agent 规划能力：ReAct 模式

ReAct（Reasoning + Acting）是一种经典的 Agent 规划模式，它让模型交替进行“思考”和“行动”，从而解决复杂问题。LangChain 的 create_react_agent 实现了这一模式。

from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
react_prompt = PromptTemplate.from_template(
"""你是一个智能助手，请按照以下步骤解决问题：
问题：{input}
请逐步思考：
分析问题需要哪些信息
决定调用哪个工具
根据工具结果继续推理
可用工具：
{tools}
工具名称列表：{tool_names}
思考过程：{agent_scratchpad}
"""
)
react_agent = create_react_agent(llm, tools, react_prompt)
react_executor = AgentExecutor(
agent=react_agent,
tools=tools,
verbose=True,
max_iterations=5  # 限制最大迭代次数，防止死循环
)
result = react_executor.invoke({
"input": "计算 (15 + 27) * 3 的结果，并查询上海的天气"
})
print(result["output"])

ReAct 模式的核心优势在于可解释性：每一步推理过程都可以被追踪和审计，便于调试和优化。

7. 实战：构建一个数据分析 Agent

下面综合运用前面所学，构建一个能够读取 CSV 文件、执行数据分析并生成报告的多工具 Agent。

import pandas as pd
import json
from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
模拟销售数据
data = {
"月份": ["1月", "2月", "3月", "4月", "5月", "6月"],
"销售额": [12000, 15000, 13000, 18000, 22000, 25000],
"订单数": [300, 380, 320, 450, 520, 600]
}
df = pd.DataFrame(data)
df.to_csv("sales_data.csv", index=False)
@tool
def load_data(file_path: str) -> str:
"""加载 CSV 数据文件并返回基本信息"""
df = pd.read_csv(file_path)
info = {
"列名": list(df.columns),
"行数": len(df),
"数据预览": df.head(3).to_dict(orient="records")
}
return json.dumps(info, ensure_ascii=False)
@tool
def calculate_statistics(column: str) -> str:
"""计算指定列的统计指标（均值、最大值、最小值）"""
df = pd.read_csv("sales_data.csv")
if column not in df.columns:
return f"列 {column} 不存在"
stats = {
"列": column,
"均值": float(df[column].mean()),
"最大值": float(df[column].max()),
"最小值": float(df[column].min())
}
return json.dumps(stats, ensure_ascii=False)
@tool
def find_best_month() -> str:
"""找出销售额最高的月份"""
df = pd.read_csv("sales_data.csv")
best_idx = df["销售额"].idxmax()
return json.dumps({
"最佳月份": df.loc[best_idx, "月份"],
"销售额": float(df.loc[best_idx, "销售额"])
}, ensure_ascii=False)
创建 Agent
analysis_tools = [load_data, calculate_statistics, find_best_month]
analysis_llm = ChatOpenAI(model="gpt-4o", api_key=api_key, base_url=base_url)
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个数据分析助手，使用工具分析销售数据并给出专业结论。"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(analysis_llm, analysis_tools, prompt)
executor = AgentExecutor(agent=agent, tools=analysis_tools, verbose=True)
执行分析任务
result = executor.invoke({
"input": "请分析 sales_data.csv 的销售数据，告诉我哪个月销售额最高，以及整体销售趋势如何？"
})
print(result["output"])

这个 Agent 能够自主完成数据加载、统计计算、趋势分析，并生成一份包含结论的自然语言报告。

8. 部署与最佳实践

将 Agent 部署到生产环境时，需要注意以下几个方面：

8.1 使用 FastAPI 封装 API

pip install fastapi uvicorn

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
app = FastAPI(title="AI Agent API")
class AgentRequest(BaseModel):
input: str
session_id: str = "default"
class AgentResponse(BaseModel):
output: str
全局 Agent 执行器（生产环境建议使用连接池）
agent_executor = None
def get_executor():
global agent_executor
if agent_executor is None:
# 初始化 Agent（复用前面的代码）
agent_executor = create_agent_executor()
return agent_executor
@app.post("/agent", response_model=AgentResponse)
async def run_agent(request: AgentRequest):
try:
executor = get_executor()
result = executor.invoke({"input": request.input})
return AgentResponse(output=result["output"])
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
return {"status": "ok"}
if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)

8.2 最佳实践清单

• 限制迭代次数：设置 max_iterations 防止 Agent 陷入死循环。
• 错误处理：为每个工具添加异常捕获，返回友好的错误信息。
• 日志记录：记录每次工具调用的输入输出，便于审计和调试。
• 安全控制：对工具权限进行严格管理，避免 Agent 执行危险操作。
• 成本控制：使用 max_tokens 限制输出长度，合理选择模型规格。
• 缓存机制：对频繁查询的结果进行缓存，降低 API 调用成本。
• 并发处理：使用异步框架（如 FastAPI）处理高并发请求。

9. 总结与进阶方向

本文从零开始介绍了 AI Agent 的核心概念、开发环境和实战案例。通过工具调用、记忆管理、规划模式和部署实践，你已经掌握了构建一个生产级 Agent 的基础能力。

接下来可以继续探索以下进阶方向：

• 多 Agent 协作：使用 LangGraph 构建多个 Agent 协同工作的复杂系统。
• RAG 增强：结合检索增强生成，让 Agent 访问私有知识库。
• 多模态 Agent：支持图像、音频、视频等多模态输入输出。
• 自主规划框架：研究 AutoGPT、BabyAGI 等自主 Agent 的规划策略。
• 评估与优化：建立 Agent 的自动化评估体系，持续优化提示词和工具设计。

AI Agent 是当前人工智能领域最具潜力的方向之一，掌握这项技术将为你的职业发展打开新的空间。建议从简单的工具调用开始，逐步构建复杂的多工具、多记忆、多规划能力的 Agent 系统，在实践中不断积累经验。