在技术社区里我们经常讨论如何利用像 ChatGPT 这样的通用大语言模型LLM来辅助编程、解答问题或生成文本。然而一个正在发生的深刻转变是那些构建了这些强大模型的核心团队其工作模式已经超越了“人机对话”的初级阶段。他们不再仅仅将 ChatGPT 视为一个问答工具而是将其作为核心引擎驱动一系列能够自主感知、规划、决策和执行的“智能体”AI Agent。这种从“使用工具”到“构建智能体”的范式迁移正在重新定义软件开发和系统运维的边界。对于开发者而言理解并实践 AI Agent 的开发不再是追赶潮流而是构建下一代人机协作系统的核心技能。本文将从一个工程实践者的视角带你理解 AI Agent 的核心概念并基于 OpenAI 的 Codex 模型或其兼容替代方案如 DeepSeek和 Agent 开发框架完成一个从零搭建、可运行、可扩展的 AI 智能体项目。你将了解到如何让 AI 不只是回答问题而是能理解复杂指令、调用工具、处理数据并完成一个完整的工作流。1. 理解 AI Agent从对话模型到自主执行体在深入代码之前我们必须先厘清几个核心概念这决定了我们构建的究竟是一个“高级脚本”还是一个真正的“智能体”。1.1 什么是 AI Agent通俗地讲一个 AI Agent 是一个能够感知环境、自主决策并执行行动以实现特定目标的软件实体。它不同于传统的程序需要明确的每一步指令也不同于基础的聊天机器人仅进行对话。它的核心在于“自主性”和“目标导向性”。技术定义上一个典型的 AI Agent 系统通常包含以下核心组件规划模块Planner将用户的高层目标分解为一系列可执行的任务或步骤。记忆模块Memory存储对话历史、任务状态、工具调用结果和世界知识为决策提供上下文。工具使用模块Tool Use能够调用外部函数、API、数据库或命令行工具来获取信息或改变环境状态。行动执行模块Actuator实际执行工具调用并处理返回结果。反思与学习模块Reflection评估行动结果修正错误优化后续策略。1.2 为什么“造 ChatGPT 的人”转向了 AgentOpenAI 的 Codex 模型GPT-3 的代码版本和后续的 GPT 系列其训练数据包含了海量的代码和自然语言。这使得它们非常擅长理解意图和生成代码片段。然而直接使用 ChatGPT 交互存在局限状态不连续每次对话相对独立难以维持复杂的、多步骤的任务状态。缺乏执行力它知道“如何做”但无法“亲自去做”比如无法直接运行它生成的 SQL 去查询数据库。信息滞后它的知识截止于训练数据无法获取实时信息如当前天气、股票价格或私有数据如公司内部数据库。因此将 LLM 作为 Agent 的“大脑”为其配备“记忆”向量数据库、“手脚”工具函数和“反思”能力就能构建出能够处理开放式、多步骤复杂任务的系统。这正是当前 AI 工程的前沿。1.3 关键术语关联ChatGPT, Codex, OpenAI API, DeepSeekChatGPT基于 GPT 系列模型的对话式应用产品提供了友好的 Web 界面和对话能力。CodexOpenAI 专门为代码生成和代码理解优化的模型系列是 GitHub Copilot 背后的引擎。它理解代码上下文的能力极强。OpenAI API提供编程接口允许开发者调用包括 GPT-4, GPT-3.5-Turbo, Codex 等在内的模型。构建 Agent 通常通过 API 与模型“大脑”交互。DeepSeek一个强大的开源 LLM其最新版本如 DeepSeek-V3在代码和推理能力上表现优异并且提供了与 OpenAI API兼容的接口。这意味着你可以用更低的成本将原本为 OpenAI API 设计的 Agent 框架几乎无缝地迁移到 DeepSeek 上这对于学习和生产部署都极具价值。注意本文的示例将基于OpenAI API 兼容格式进行设计。这意味着你可以选择使用 OpenAI 的官方端点也可以使用 DeepSeek 等提供的兼容端点只需替换base_url和api_key即可。2. 环境准备与项目初始化我们将构建一个简单的“数据分析智能体”。它的目标是接收用户关于数据的自然语言问题例如“上个月销售额最高的产品是什么”自动编写并执行相应的 SQL 查询然后对查询结果进行总结以自然语言形式回复给用户。2.1 技术栈与工具选择为了快速构建原型我们选择以下技术栈Python 3.9 主要的开发语言。LangChain / LangGraph 当前最流行的 Agent 开发框架之一提供了构建链Chain、代理Agent和图Graph的高层抽象。我们将使用其较新的LangGraph来构建有状态的 Agent。OpenAI SDK / 兼容 SDK 用于调用 LLM。我们将使用openai这个官方包但配置为指向兼容端点。SQLite 作为示例数据库轻量且无需额外服务。Chroma 轻量级向量数据库用于实现 Agent 的短期记忆本例中可能简化但展示概念。2.2 创建项目与安装依赖首先创建一个新的项目目录并初始化虚拟环境。mkdir ai-data-agent cd ai-data-agent python -m venv venv # 在 Windows 上激活 venv\Scripts\activate # 在 macOS/Linux 上激活 source venv/bin/activate接下来创建requirements.txt文件并安装依赖。# requirements.txt langchain0.1.0 langchain-openai0.0.5 langgraph0.0.29 openai1.12.0 chromadb0.4.22 sqlalchemy2.0.25 pandas2.1.4 python-dotenv1.0.0安装命令pip install -r requirements.txt2.3 配置 API 密钥与环境变量为了安全地管理 API 密钥我们使用.env文件。在项目根目录创建.env文件。# .env # 如果你使用 OpenAI 官方服务 OPENAI_API_KEYsk-your-openai-api-key-here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果你使用 DeepSeek 兼容服务 # DEEPSEEK_API_KEYyour-deepseek-api-key-here # DEEPSEEK_BASE_URLhttps://api.deepseek.com/v1重要永远不要将 API 密钥硬编码在代码中或提交到版本控制系统如 Git。确保.env文件已被添加到.gitignore中。创建一个config.py文件来读取配置。# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 # 优先使用 DeepSeek 配置如果未设置则回退到 OpenAI api_key os.getenv(DEEPSEEK_API_KEY) or os.getenv(OPENAI_API_KEY) base_url os.getenv(DEEPSEEK_BASE_URL) or os.getenv(OPENAI_BASE_URL) if not api_key: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY 或 DEEPSEEK_API_KEY) # 使用的模型名称根据你的端点支持情况调整 # OpenAI: gpt-3.5-turbo, gpt-4-turbo-preview # DeepSeek: deepseek-chat MODEL_NAME deepseek-chat if deepseek in base_url else gpt-3.5-turbo print(f配置加载成功: 使用端点 {base_url}, 模型 {MODEL_NAME})3. 构建核心组件数据库、工具与 Agent 大脑一个能执行 SQL 的 Agent 需要几个核心部分一个真实的数据库、操作数据库的工具函数以及一个能协调这些工具的“大脑”。3.1 创建示例数据库与数据我们创建一个 SQLite 数据库并插入一些模拟的销售数据。创建init_database.py脚本。# init_database.py import sqlite3 import pandas as pd from datetime import datetime, timedelta # 连接数据库如果不存在则创建 conn sqlite3.connect(sales_data.db) cursor conn.cursor() # 创建销售记录表 cursor.execute( CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_name TEXT NOT NULL, category TEXT, sale_date DATE NOT NULL, quantity INTEGER NOT NULL, unit_price REAL NOT NULL, region TEXT ) ) # 生成模拟数据 products [Laptop, Mouse, Keyboard, Monitor, Webcam] categories [Electronics, Accessories, Electronics, Electronics, Accessories] regions [North, South, East, West] data [] start_date datetime.now() - timedelta(days60) for i in range(100): product_idx i % len(products) sale_date start_date timedelta(daysi % 30, hours(i*3)%24) row ( products[product_idx], categories[product_idx], sale_date.date().isoformat(), (i % 5) 1, # 数量 1-5 [799.99, 25.50, 89.99, 299.99, 45.00][product_idx], regions[i % len(regions)] ) data.append(row) # 插入数据 cursor.executemany( INSERT INTO sales (product_name, category, sale_date, quantity, unit_price, region) VALUES (?, ?, ?, ?, ?, ?) , data) conn.commit() # 验证数据 df pd.read_sql_query(SELECT * FROM sales LIMIT 5, conn) print(数据库表 sales 创建成功示例数据) print(df) print(f\n总记录数: {pd.read_sql_query(SELECT COUNT(*) as count FROM sales, conn)[count][0]}) conn.close()运行这个脚本以初始化数据库python init_database.py3.2 定义 Agent 可用的工具Tools工具是 Agent 与外界交互的桥梁。我们定义一个执行 SQL 查询的工具。创建tools.py文件。# tools.py from langchain.tools import tool import sqlite3 import pandas as pd import json tool def query_database(query: str) - str: 执行一个 SQL 查询语句并返回结果。 参数: query: 一个有效的 SQL SELECT 查询字符串。 返回: 一个格式化的字符串包含查询结果或错误信息。 try: conn sqlite3.connect(sales_data.db) # 使用 pandas 方便地读取和格式化结果 df pd.read_sql_query(query, conn) conn.close() if df.empty: return 查询成功但未返回任何数据。 else: # 将 DataFrame 转换为更易读的格式例如 Markdown 表格字符串 # 也可以返回 JSON但字符串更适合 LLM 理解 result_str df.to_markdown(indexFalse) return f查询成功返回 {len(df)} 行数据\n\n{result_str}\n except sqlite3.Error as e: return f数据库查询错误: {e} except Exception as e: return f执行查询时发生未知错误: {e} # 工具列表可供 Agent 使用 available_tools [query_database]这个tool装饰器来自 LangChain它能够自动生成符合 OpenAI Function Calling 格式的工具描述这对于 Agent 理解工具用途至关重要。3.3 构建 Agent 的“大脑”与工作流我们将使用 LangGraph 来定义 Agent 的状态和决策循环。创建agent_graph.py文件。# agent_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import Tool from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from config import api_key, base_url, MODEL_NAME from tools import available_tools # 1. 定义 Agent 的状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话消息历史 query_result: str # 存储工具调用结果 # 2. 初始化 LLM并为其绑定工具 llm ChatOpenAI( modelMODEL_NAME, api_keyapi_key, base_urlbase_url, temperature0 # 对于执行任务低温度更稳定 ) # 将工具绑定到 LLM使其具备调用能力 llm_with_tools llm.bind_tools(available_tools) # 3. 定义节点函数 def call_model(state: AgentState): 调用 LLM决定下一步是回复用户还是调用工具。 messages state[messages] response llm_with_tools.invoke(messages) # 将 LLM 的响应添加到消息历史中 return {messages: [response]} def execute_tools(state: AgentState): 执行 LLM 要求调用的工具。 messages state[messages] last_message messages[-1] tool_calls last_message.tool_calls if not tool_calls: raise ValueError(没有需要执行的工具调用) tool_messages [] for tool_call in tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 根据工具名找到对应的工具函数 tool_to_use next((t for t in available_tools if t.name tool_name), None) if not tool_to_use: tool_result f错误未知工具 {tool_name} else: # 实际执行工具 tool_result tool_to_use.invoke(tool_args) # 创建工具执行结果消息 tool_messages.append(ToolMessage( contentstr(tool_result), tool_call_idtool_call[id] )) # 将工具执行结果添加到状态 return {messages: tool_messages, query_result: str(tool_result)} # 4. 定义条件路由函数 def should_continue(state: AgentState) - str: 根据最后一个消息判断下一步是调用工具还是结束。 messages state[messages] last_message messages[-1] # 如果 LLM 返回了工具调用请求则去执行工具 if last_message.tool_calls: return call_tool # 否则对话结束 return end # 5. 构建图工作流 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) # Agent 思考节点 workflow.add_node(action, execute_tools) # 工具执行节点 # 设置入口点 workflow.set_entry_point(agent) # 添加条件边 workflow.add_conditional_edges( agent, should_continue, { call_tool: action, # 需要调用工具则跳转到 action 节点 end: END # 否则结束 } ) # 工具执行完后继续让 Agent 思考 workflow.add_edge(action, agent) # 编译图 app workflow.compile()这个图定义了一个经典的 ReAct (Reasoning Acting) 循环Agent节点接收当前状态对话历史让 LLM 思考。LLM 可以决定直接回复用户或者请求调用一个工具如query_database。条件路由检查 LLM 的输出。如果有工具调用则前往action节点如果没有则工作流结束。Action节点执行 LLM 请求调用的具体工具并将执行结果作为消息返回。循环工具执行结果被添加回消息历史后流程回到agent节点LLM 根据新结果再次思考直到它认为任务完成并给出最终回答。4. 运行与验证智能体现在我们将智能体组装起来并通过几个不同复杂度的问题来验证其能力。4.1 创建主程序并测试创建main.py作为程序的入口点。# main.py from agent_graph import app from langchain_core.messages import HumanMessage import asyncio async def run_agent(user_query: str): 运行智能体处理用户查询 print(f\n用户问题: {user_query}) print(- * 50) # 初始化状态包含用户的问题 initial_state { messages: [HumanMessage(contentuser_query)], query_result: } final_state None # 我们逐步执行图以便观察中间过程 async for event in app.astream(initial_state): for node_name, node_output in event.items(): if node_name ! __end__: last_message node_output.get(messages, [])[-1] if last_message: # 打印 LLM 的思考或工具调用请求 if hasattr(last_message, tool_calls) and last_message.tool_calls: print(f[Agent 决定调用工具]) for tc in last_message.tool_calls: print(f 工具: {tc[name]}) print(f 参数: {tc[args]}) elif last_message.content: # 打印 LLM 的最终回答或中间推理 print(f[Agent 回复]: {last_message.content}) # 记录最终状态 final_state node_output print(- * 50) return final_state if __name__ __main__: # 测试用例 test_queries [ 上个月总销售额是多少, # 需要计算 销量最好的产品是哪一款, # 需要聚合和排序 给我看看最近10条销售记录。, # 简单查询 对比一下 North 和 South 地区的平均销售额。, # 复杂查询和对比 ] async def main(): for query in test_queries: await run_agent(query) await asyncio.sleep(1) # 简单间隔避免请求过快 asyncio.run(main())运行程序观察智能体的思考过程python main.py4.2 预期输出与过程解析运行上述测试你应该能看到类似以下的输出具体 SQL 和数字可能因数据随机生成而不同用户问题: 上个月总销售额是多少 -------------------------------------------------- [Agent 决定调用工具] 工具: query_database 参数: {query: SELECT SUM(quantity * unit_price) as total_sales FROM sales WHERE sale_date date(now, -1 month)} [Agent 回复]: 根据查询结果上个月过去30天的总销售额为 12,345.67 美元。 --------------------------------------------------过程解析用户输入自然语言问题。AgentLLM接收到消息理解到需要计算“上个月总销售额”。LLM 知道它有一个query_database工具并自主生成了合适的 SQL 查询语句。它正确地处理了日期逻辑date(now, -1 month)和销售额计算quantity * unit_price。工作流将执行权交给action节点该节点调用query_database工具运行生成的 SQL。工具返回查询结果一个包含总销售额的 Markdown 表格字符串。这个结果被作为ToolMessage添加回对话历史。工作流再次进入agent节点LLM 看到工具返回的具体数据然后生成最终的自然语言回复给用户。对于更复杂的问题“对比一下 North 和 South 地区的平均销售额”Agent 可能会生成更复杂的 SQL如SELECT region, AVG(quantity * unit_price) as avg_sales FROM sales WHERE region IN (North, South) GROUP BY region并最终给出对比结论。5. 核心机制详解与关键参数5.1 LangGraph 状态管理AgentState是一个TypedDict它定义了智能体工作流中流动的数据结构。Annotated[List, operator.add]是一个特殊的注解告诉 LangGraphmessages字段在流经不同节点时应该用列表相加的方式合并从而自然地累积对话历史。这是实现 Agent “记忆”的基础。5.2 工具绑定与 Function Callingllm.bind_tools(available_tools)是至关重要的一步。它将工具的函数签名名称、描述、参数 schema以特定的格式OpenAI Function Calling注入到 LLM 的上下文中。这使得 LLM 在生成回复时可以选择输出一个特殊的“工具调用请求”结构而不是普通文本。框架会解析这个结构并路由到对应的工具函数。5.3 模型参数与稳定性在初始化ChatOpenAI时我们设置了temperature0。对于执行确定任务的 Agent较低的 temperature接近 0可以使模型输出更加确定和一致减少随机性这对于生成准确的 SQL 语句至关重要。在创意性任务中可以适当调高。5.4 工作流循环与停止条件should_continue函数是工作流的“调度器”。它检查 LLM 的最后一条消息如果包含tool_calls则路由到action节点。如果不包含则路由到END工作流终止。 这个简单的逻辑构成了 Agent 的自主决策循环思考 - 行动 - 观察 - 再思考直到任务完成。6. 常见问题排查与优化在实际开发和运行中你可能会遇到以下问题。6.1 Agent 无法正确生成 SQL问题现象可能原因检查与解决LLM 回复说“我不会写 SQL”或生成非 SQL 文本。1. 模型能力不足。2. 系统提示词System Prompt不明确。3. 工具描述不够清晰。1.升级模型尝试使用能力更强的模型如 GPT-4 或 DeepSeek 的最新版本。2.增强提示在初始的HumanMessage前可以添加一个SystemMessage明确角色和任务。例如SystemMessage(content你是一个数据分析专家擅长将自然语言问题转化为精确的 SQL 查询。你只能使用提供的工具来查询数据库。)。3.优化工具描述确保tool装饰器下的函数文档字符串清晰描述了工具的用途和输入格式。生成的 SQL 语法错误执行失败。1. LLM 对数据库 schema 不了解。2. 复杂逻辑理解有偏差。1.提供 Schema 上下文在系统提示词或首次用户消息中提供数据库表的 DDL 信息。例如“数据库有一个sales表包含字段id, product_name, category, sale_date, quantity, unit_price, region。”2.迭代优化收集出错的 SQL 案例将其和正确写法作为 few-shot 示例加入提示词中引导模型学习。SQL 查询结果为空但预期有数据。1. 查询条件错误如日期范围。2. 字段名或表名拼写错误。1.让 Agent 反思在工具函数query_database返回空结果时可以返回更详细的提示如“查询成功但结果为空请检查查询条件如日期、产品名是否正确。”这会被 LLM 看到并用于下一轮思考。2.添加验证在工具调用前可以添加一个简单的 SQL 语法或关键词校验层但注意不要过度限制。6.2 工作流陷入无限循环问题现象可能原因检查与解决Agent 反复调用同一个工具无法给出最终答案。1. LLM 无法从工具结果中提炼出最终答案。2. 停止条件判断逻辑有缺陷。1.增强 LLM 的总结能力在系统提示词中强调“在获得数据后你需要用简洁的语言向用户总结核心发现”。2.设置最大迭代次数这是生产环境必备的防护措施。可以在should_continue函数中添加一个计数器到state中当超过阈值如10次时强制返回end。LangGraph 也支持在编译图时设置interrupt_before或interrupt_after来管理循环。6.3 性能与成本优化关注点潜在问题优化建议响应延迟每个思考-行动步骤都需要调用一次 LLM API复杂任务步骤多总延迟高。1.使用流式响应如示例所用astream可以让用户尽早看到部分输出。2.优化提示词清晰的指令可以减少 LLM 的“困惑”和无效输出缩短完成任务的步数。3.并行工具调用如果多个工具调用间无依赖LangGraph 支持并行执行。API 调用成本每一步都消耗 Token多轮对话成本累积。1.选择性价比模型对于工具调用等任务gpt-3.5-turbo或deepseek-chat通常足够成本远低于 GPT-4。2.压缩历史当对话历史很长时可以总结之前的交互而非传递全部原始消息减少 Token 消耗。这属于“记忆管理”的高级话题。3.设置预算上限在客户端监控 Token 消耗设置硬性上限。工具执行安全Agent 生成的 SQL 可能是DELETE或DROP等危险操作。1.工具层面限制在query_database工具函数中在执行前检查 SQL 语句只允许SELECT开头的查询语句拒绝其他 DML/DDL 操作。2.数据库权限隔离为 Agent 使用的数据库连接设置只读权限。7. 从原型到生产最佳实践与扩展方向7.1 生产环境清单将上述原型部署到生产环境至少需要考虑以下方面配置管理将 API 密钥、模型名称、数据库连接字符串等全部移至环境变量或配置中心。错误处理与重试为 LLM API 调用和工具调用添加完善的错误处理如网络超时、速率限制和指数退避重试机制。日志与监控记录完整的 Agent 执行轨迹包括每步的输入、LLM 响应、工具调用及结果。这对于调试和优化至关重要。监控 Token 消耗、请求延迟和错误率。权限与安全如前所述严格限制工具的执行权限。对用户输入进行基本的清洗和校验防止提示词注入攻击。状态持久化当前的AgentState在内存中。对于 Web 服务需要将会话状态messages持久化到数据库或 Redis 中并关联用户会话 ID。版本控制对 Agent 的工作流定义、提示词、工具集进行版本控制。7.2 扩展智能体能力一个数据分析 Agent 只是起点。你可以通过增加工具来扩展其能力数据可视化工具添加一个generate_chart工具接收数据和分析要求调用 Matplotlib 或 Plotly 生成图表并保存为图片或返回 Base64 编码。文件处理工具添加read_csv、analyze_excel_sheet工具让 Agent 能处理用户上传的文件。网络搜索工具集成 SerperAPI 或 Tavily 搜索让 Agent 能获取实时信息来补充分析。代码执行工具沙盒环境在安全的沙盒中执行 Python 代码片段进行更复杂的数据转换或计算。多 Agent 协作使用 LangGraph 的更高阶功能创建多个 specialized Agent如一个负责 SQL 生成一个负责结果解读一个负责报告撰写让它们通过共享状态协同工作。7.3 学习路径建议要深入掌握 AI Agent 开发建议按以下路径推进巩固基础深入理解 LangChain/LangGraph 的核心概念——Model I/O, Chains, Agents, Tools, Memory。掌握工具调用熟练创建各种类型的工具并理解 OpenAI Function Calling 的底层协议。设计复杂工作流学习使用 LangGraph 构建带条件分支、循环、并行、子图的工作流以处理复杂、多阶段的任务。优化提示工程学习编写有效的系统提示词、少样本示例以及进行思维链Chain-of-Thought提示以提升 Agent 的推理质量。集成向量数据库为 Agent 添加长期记忆使其能记住过去对话的关键信息并能从知识库中检索相关文档。探索多模态尝试让 Agent 处理图像、音频输入或生成多模态输出。研究 Agentic 框架关注 AutoGen, CrewAI 等其他框架了解不同的 Agent 设计范式。从“使用 ChatGPT 聊天”到“构建自主工作的 AI Agent”是开发者利用大模型能力的一次关键跃迁。这个过程要求我们不仅会调用 API更要具备系统思维将 LLM 视为一个具有规划和推理能力的核心组件并为其设计可靠的环境感知与动作执行闭环。本文通过一个可运行的数据分析 Agent 示例展示了从环境搭建、工具定义、工作流构建到问题排查的完整路径。真正的挑战和乐趣在于根据你所要解决的具体领域问题设计出恰到好处的工具集和工作流让 AI 智能体成为你业务中高效、可靠的数字员工。
