AI Agent 工程师入门指南：从原理到实战

1. 什么是 AI AgentAI Agent智能体是一种能够感知环境、做出决策并执行动作的智能系统。与传统的大语言模型LLM不同Agent 不仅仅是生成文本它能够调用工具、访问外部数据、规划任务步骤并在多轮交互中自主完成复杂目标。一个典型的 AI Agent 通常包含以下核心组件大语言模型LLM作为推理和决策的“大脑”负责理解用户意图、生成计划和回复。工具调用Tool CallingAgent 通过调用外部 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_KEYsk-your-api-key-here OPENAI_BASE_URLhttps://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_asciiFalse) return json.dumps({错误: f未找到城市 {city} 的天气数据}, ensure_asciiFalse)然后定义工具描述并让模型决定何时调用from openai import OpenAI client OpenAI(api_keyapi_key, base_urlbase_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( modelgpt-4o, messagesmessages, toolstools, tool_choiceauto ) 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( modelgpt-4o, messagesmessages ) return final_response.choices[0].message.content return assistant_message.content 测试 print(run_agent(北京今天天气怎么样))运行上述代码Agent 会自动识别需要调用get_weather工具获取天气数据后生成自然语言回复。4. 使用 LangChain 构建 AgentLangChain 提供了更高层的抽象让 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( modelgpt-4o, api_keyapi_key, base_urlbase_url, temperature0 ) 定义工具 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(agentagent, toolstools, verboseTrue) 运行 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_keychat_history, return_messagesTrue ) prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}) ]) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue ) 第一轮对话 agent_executor.invoke({input: 我叫小明请记住我的名字。}) 第二轮对话Agent 能记住用户名字 result agent_executor.invoke({input: 我叫什么名字}) print(result[output])5.2 长期记忆向量数据库对于跨会话的长期记忆可以使用向量数据库存储和检索信息。下面使用 Chroma 作为向量存储pip install chromadb langchain-communityfrom langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import Chroma from langchain.text_splitter import CharacterTextSplitter 初始化嵌入模型 embeddings OpenAIEmbeddings( api_keyapi_key, base_urlbase_url, modeltext-embedding-3-small ) 创建向量数据库 vectorstore Chroma( collection_nameagent_memory, embedding_functionembeddings, 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, kk) return [doc.page_content for doc in results] 示例保存用户偏好 save_memory(用户小明喜欢简洁的回答风格偏好 Python 技术栈。) save_memory(用户小明的项目是电商平台使用 Django 框架。) 检索记忆 memories retrieve_memory(用户喜欢什么技术栈) print(检索到的记忆:, memories)6. Agent 规划能力ReAct 模式ReActReasoning 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( agentreact_agent, toolstools, verboseTrue, max_iterations5 # 限制最大迭代次数防止死循环 ) 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, indexFalse) 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(orientrecords) } return json.dumps(info, ensure_asciiFalse) 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_asciiFalse) 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_asciiFalse) 创建 Agent analysis_tools [load_data, calculate_statistics, find_best_month] analysis_llm ChatOpenAI(modelgpt-4o, api_keyapi_key, base_urlbase_url) prompt ChatPromptTemplate.from_messages([ (system, 你是一个数据分析助手使用工具分析销售数据并给出专业结论。), (human, {input}), (placeholder, {agent_scratchpad}) ]) agent create_tool_calling_agent(analysis_llm, analysis_tools, prompt) executor AgentExecutor(agentagent, toolsanalysis_tools, verboseTrue) 执行分析任务 result executor.invoke({ input: 请分析 sales_data.csv 的销售数据告诉我哪个月销售额最高以及整体销售趋势如何 }) print(result[output])这个 Agent 能够自主完成数据加载、统计计算、趋势分析并生成一份包含结论的自然语言报告。8. 部署与最佳实践将 Agent 部署到生产环境时需要注意以下几个方面8.1 使用 FastAPI 封装 APIpip install fastapi uvicornfrom fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn app FastAPI(titleAI 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_modelAgentResponse) async def run_agent(request: AgentRequest): try: executor get_executor() result executor.invoke({input: request.input}) return AgentResponse(outputresult[output]) except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): return {status: ok} if name main: uvicorn.run(app, host0.0.0.0, port8000)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 系统在实践中不断积累经验。