从零构建AI代理：基于LangChain实现工具调用与记忆机制

在实际 AI 应用开发中构建一个能够理解上下文、自主执行任务并持续学习的智能代理AI Agent是许多开发者和创业团队探索的方向。近期一个名为“Grok Bot”的 AI 代理项目因其宣称的“一人团队”运营模式而受到关注其核心思路是利用大型语言模型LLM作为大脑结合工具调用和记忆机制模拟一个微型公司的决策与执行流程。对于技术实践者而言这背后的价值不在于概念本身而在于如何将“AI 代理”从一个抽象名词落地为可运行、可调试、可扩展的代码系统。本文将带你从零开始构建一个具备基础能力的 AI 代理原型。我们将聚焦于技术实现涵盖核心概念、环境搭建、关键模块编码、运行验证以及生产级考量的全过程。通过本文你将掌握如何设计一个能理解目标、规划步骤、使用工具并记忆上下文的 AI 代理并理解在将其投入实际业务前必须解决的稳定性、成本与安全问题。1. 理解 AI 代理的核心机制从 LLM 到自主执行在讨论具体代码之前必须厘清“AI 代理”与普通聊天机器人的本质区别。一个简单的问答机器人是“刺激-反应”模式而 AI 代理则引入了“目标-规划-执行-反思”的循环。1.1 代理与工具赋予模型“手脚”大型语言模型本身是强大的推理和文本生成引擎但它无法直接操作外部系统如数据库、API、文件。AI 代理通过“工具Tools”机制解决了这个问题。你可以将工具理解为模型可以调用的函数。例如一个“获取天气”的工具代理在判断需要天气信息时会生成调用该工具所需的参数系统执行后再将结果返回给代理进行后续分析。这构成了代理与外界交互的基本单元。1.2 规划与执行分解复杂任务面对“分析上周销售数据并生成报告”这样的复杂指令代理不会试图一步完成。它会进行任务分解Planning例如1. 调用工具获取销售数据2. 调用工具进行数据清洗3. 调用分析工具计算关键指标4. 调用报告生成工具。这个过程可以是链式的也可以根据中间结果动态调整。1.3 记忆与上下文维持对话状态短期记忆Short-term Memory通常指当前会话的上下文窗口模型能记住最近的对话历史。长期记忆Long-term Memory则涉及将关键信息持久化存储如向量数据库供后续会话检索。这对于实现“一人团队”的持续性至关重要代理需要记住过去的决策、用户的偏好和已完成的任务。1.4 反思与迭代从错误中学习高级代理会引入“反思Reflection”环节。当任务执行失败或结果不理想时代理可以分析错误日志、工具输出重新评估计划并尝试替代方案。这模拟了人类从试错中学习的过程。理解了这些核心组件我们就能将它们映射到具体的代码模块上。接下来我们将选择一个流行的开发框架来搭建我们的原型。2. 环境准备与框架选型构建 AI 代理不需要从零发明轮子利用成熟的开发框架可以极大提升效率。这里我们选择LangChain作为核心框架因为它提供了丰富的工具集成、记忆模块和代理执行器生态活跃文档齐全。2.1 基础环境与依赖首先确保你的开发环境已安装 Python推荐 3.9 及以上版本。我们将使用虚拟环境来管理依赖。# 创建并激活虚拟环境以 Linux/macOS 为例 python -m venv ai_agent_env source ai_agent_env/bin/activate # 对于 Windows # ai_agent_env\Scripts\activate接下来安装核心依赖。除了langchain我们还需要一个 LLM 提供商这里以 OpenAI 为例但你也可以选择其他兼容 API、处理环境变量的python-dotenv以及用于向量记忆的chromadb。pip install langchain langchain-openai python-dotenv chromadb tiktokentiktoken是 OpenAI 用于计算 Token 的库对于成本估算和上下文窗口管理有帮助。2.2 配置 LLM 连接为了调用 LLM如 GPT-3.5-Turbo 或 GPT-4你需要一个 API 密钥。强烈建议将密钥存储在环境变量中而不是硬编码在代码里。创建一个名为.env的文件在项目根目录OPENAI_API_KEY你的_OpenAI_API_密钥_在这里然后在 Python 代码中加载并使用它# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY 环境变量)2.3 项目结构规划一个清晰的项目结构有助于后续维护和扩展。建议按以下方式组织ai_agent_project/ ├── .env # 环境变量切勿提交至版本控制 ├── .gitignore # Git忽略文件 ├── requirements.txt # 项目依赖 ├── main.py # 主程序入口 ├── config.py # 配置管理 ├── agents/ # 代理相关模块 │ ├── __init__.py │ ├── base_agent.py # 代理基类或核心逻辑 │ └── specialized_agent.py # 特定领域代理 ├── tools/ # 工具定义 │ ├── __init__.py │ ├── calculator.py # 示例计算器工具 │ ├── web_search.py # 示例网络搜索工具需额外API │ └── database_tool.py # 示例数据库查询工具 ├── memory/ # 记忆模块 │ ├── __init__.py │ ├── short_term.py # 短期记忆管理 │ └── long_term.py # 长期记忆向量存储 └── utils/ # 工具函数 ├── __init__.py └── logger.py # 日志配置现在环境和项目骨架已经准备就绪。接下来我们将从最核心的部分——定义工具开始。3. 实现核心模块工具、记忆与代理我们将采用自底向上的方式先实现代理赖以生存的工具和记忆系统最后组装成完整的代理。3.1 创建自定义工具一个工具本质上是一个能被 LangChain 代理识别的函数。我们创建一个简单的计算器工具和一个模拟获取用户信息的工具。# tools/calculator.py from langchain.tools import tool import math tool def calculator(expression: str) - str: 执行一个数学表达式计算。支持加减乘除和括号。 例如calculator((3 5) * 2) 将返回 16。 注意出于安全考虑请勿直接使用 eval 解析不可信输入。此处为示例简化。 try: # 警告在生产环境中应对表达式进行严格的安全检查和过滤 # 此处仅为演示直接使用 eval。 result eval(expression, {__builtins__: None}, {math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} # tools/user_info_tool.py from langchain.tools import tool import datetime tool def get_user_profile(user_id: str) - str: 根据用户ID获取模拟的用户资料。 在实际项目中这里会连接数据库或用户服务。 # 模拟数据 mock_db { user_001: {name: 张三, join_date: 2023-01-15, level: VIP}, user_002: {name: 李四, join_date: 2023-06-22, level: 普通}, } user_data mock_db.get(user_id) if user_data: return f用户信息: 姓名 {user_data[name]}, 注册于 {user_data[join_date]}, 等级 {user_data[level]}. else: return f未找到用户ID为 {user_id} 的信息。tool装饰器会告诉 LangChain 这是一个可用的工具并自动从函数文档字符串docstring中提取描述供 LLM 理解工具用途。3.2 配置短期与长期记忆短期记忆通常由ConversationBufferMemory或ConversationSummaryMemory管理。长期记忆则常用向量存储Vector Store来保存和检索文本片段。# memory/short_term.py from langchain.memory import ConversationBufferMemory def get_conversation_memory(): 创建一个对话缓冲记忆。 它会保存最近的对话历史并自动注入到给LLM的提示词中。 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) return memory # memory/long_term.py from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.schema import Document import os # 初始化嵌入模型和向量数据库 embeddings OpenAIEmbeddings(openai_api_keyos.getenv(OPENAI_API_KEY)) # 指定持久化目录 PERSIST_DIRECTORY ./data/chroma_db def get_vector_store(): 获取或创建向量存储实例。 # 如果目录存在则加载已有数据库 if os.path.exists(PERSIST_DIRECTORY): return Chroma(persist_directoryPERSIST_DIRECTORY, embedding_functionembeddings) else: # 否则创建一个新的空数据库 return Chroma.from_documents(documents[], embeddingembeddings, persist_directoryPERSIST_DIRECTORY) def add_to_long_term_memory(text: str, metadata: dict None): 将一段文本添加到长期记忆向量数据库。 vector_store get_vector_store() doc Document(page_contenttext, metadatametadata or {}) vector_store.add_documents([doc]) vector_store.persist() # 持久化到磁盘 def search_memory(query: str, k3): 从长期记忆中搜索相关记忆。 vector_store get_vector_store() docs vector_store.similarity_search(query, kk) return [doc.page_content for doc in docs]3.3 组装代理执行器现在我们将工具、记忆和 LLM 组合起来创建一个可以执行多步任务的代理。# agents/base_agent.py from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from tools.calculator import calculator from tools.user_info_tool import get_user_profile from memory.short_term import get_conversation_memory import os class BaseAgent: def __init__(self, model_namegpt-3.5-turbo-1106, temperature0.1): 初始化基础代理。 :param model_name: 使用的OpenAI模型。 :param temperature: 创造性越低越确定越高越随机。 self.llm ChatOpenAI(modelmodel_name, temperaturetemperature, openai_api_keyos.getenv(OPENAI_API_KEY)) self.tools [calculator, get_user_profile] # 注册工具列表 self.memory get_conversation_memory() # 定义代理的提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的AI助手。你可以使用工具来获取信息或执行计算。 如果你不知道答案就说不知道不要编造信息。 请清晰、有条理地思考。你的思考过程可以放在最后。 当前对话历史 {chat_history}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于放置工具调用和结果 (human, {input}), ]) # 创建代理 agent create_openai_tools_agent(llmself.llm, toolsself.tools, promptprompt) # 创建代理执行器它负责循环调用工具直到任务完成或达到限制 self.agent_executor AgentExecutor( agentagent, toolsself.tools, memoryself.memory, verboseTrue, # 设置为True可以看到代理的思考过程调试时非常有用 handle_parsing_errorsTrue, # 处理解析错误 max_iterations10, # 防止代理陷入死循环 ) def run(self, query: str) - str: 执行用户的查询。 try: response self.agent_executor.invoke({input: query}) return response[output] except Exception as e: return f代理执行过程中出现错误: {e}关键参数解释verboseTrue: 这是调试神器。当设置为True时控制台会输出代理的完整思考链ReAct 格式包括它决定调用哪个工具、传递什么参数、工具返回什么结果。生产环境应设为False。handle_parsing_errorsTrue: LLM 返回的工具调用格式可能不符合预期此参数能防止程序因此崩溃。max_iterations10: 限制代理的最大执行步数防止因逻辑错误或工具不可用导致无限循环消耗大量 Token。4. 运行验证与结果分析让我们编写一个主程序来测试这个基础代理。# main.py from agents.base_agent import BaseAgent from memory.long_term import add_to_long_term_memory, search_memory import sys def main(): print(初始化 AI 代理...) agent BaseAgent() # 示例1使用计算器工具 print(\n--- 测试1: 数学计算 ---) query1 “请计算 (12 34) * 2 等于多少” result1 agent.run(query1) print(f用户: {query1}) print(f代理: {result1}) # 示例2使用用户信息工具 print(\n--- 测试2: 查询用户信息 ---) query2 “用户 user_001 的资料是什么” result2 agent.run(query2) print(f用户: {query2}) print(f代理: {result2}) # 示例3结合短期记忆的连续对话 print(\n--- 测试3: 连续对话依赖短期记忆---) query3a “我刚才问的是哪个用户” result3a agent.run(query3a) print(f用户: {query3a}) print(f代理: {result3a}) # 示例4测试长期记忆 print(\n--- 测试4: 长期记忆存储与检索 ---) # 存储一些信息 add_to_long_term_memory(项目Alpha的启动会议决定下周进行第一次代码评审。, {project: Alpha, type: meeting}) add_to_long_term_memory(用户张三喜欢在晚上接收每日报告。, {user: 张三, preference: report_time}) print(已存入两条长期记忆。) # 检索信息 memories search_memory(代码评审什么时候) print(f检索到相关记忆: {memories}) # 示例5复杂任务需要规划 print(\n--- 测试5: 多步任务规划 ---) query5 “先告诉我 user_001 的等级然后计算他的等级长度字符数加上5是多少” result5 agent.run(query5) print(f用户: {query5}) print(f代理: {result5}) if __name__ __main__: main()运行python main.py你应该能看到类似以下的输出verboseTrue时更详细初始化 AI 代理... --- 测试1: 数学计算 --- 用户: 请计算 (12 34) * 2 等于多少 进入新的 AgentExecutor 链... 我需要计算这个数学表达式。 动作: calculator 动作输入: “(12 34) * 2” 观察: 计算结果: 92 思考: 我已经得到了计算结果。 最终答案: (12 34) * 2 等于 92。 链结束。 代理: (12 34) * 2 等于 92。 --- 测试2: 查询用户信息 --- 用户: 用户 user_001 的资料是什么 进入新的 AgentExecutor 链... 我需要获取用户 user_001 的资料。 动作: get_user_profile 动作输入: “user_001” 观察: 用户信息: 姓名 张三, 注册于 2023-01-15, 等级 VIP。 思考: 我已经得到了用户信息。 最终答案: 用户 user_001 的资料是姓名张三注册于2023-01-15等级VIP。 代理: 用户 user_001 的资料是姓名张三注册于2023-01-15等级VIP。 --- 测试3: 连续对话依赖短期记忆--- 用户: 我刚才问的是哪个用户 进入新的 AgentExecutor 链... 根据对话历史用户刚才询问了用户 user_001 的资料。 最终答案: 您刚才询问的是用户 user_001张三。 代理: 您刚才询问的是用户 user_001张三。 --- 测试5: 多步任务规划 --- 用户: 先告诉我 user_001 的等级然后计算他的等级长度字符数加上5是多少 进入新的 AgentExecutor 链... 我需要先获取 user_001 的等级然后计算等级字符串的长度并加5。 动作: get_user_profile 动作输入: “user_001” 观察: 用户信息: 姓名 张三, 注册于 2023-01-15, 等级 VIP。 思考: 从结果中我看到等级是“VIP”。现在计算“VIP”的长度它是3个字符然后加5。 动作: calculator 动作输入: “3 5” 观察: 计算结果: 8 思考: 计算完成。 最终答案: 用户 user_001 的等级是 VIP。VIP 的长度是3个字符加上5等于8。 代理: 用户 user_001 的等级是 VIP。VIP 的长度是3个字符加上5等于8。结果分析工具调用成功代理正确识别了需要调用get_user_profile和calculator工具并传递了正确的参数。短期记忆有效在测试3中代理无需重复询问直接从对话历史中找到了之前提到的用户。任务规划与执行在测试5中代理成功将复杂任务分解为两个顺序执行的子任务先查询后计算展示了初步的规划能力。思考链可见当verboseTrue时我们可以清晰看到代理的“思考-行动-观察”循环ReAct模式这对于调试其决策逻辑至关重要。至此一个具备基础工具调用、记忆和规划能力的 AI 代理原型已经可以运行。然而这距离一个稳定、可靠、可用于生产环境的“一人团队”还相差甚远。5. 从原型到生产关键问题与排查指南将 AI 代理投入实际应用会面临一系列在原型阶段不易察觉的挑战。以下是必须解决的几个核心问题及其排查路径。5.1 常见问题与解决方案问题现象可能原因检查与排查步骤解决方案与预防建议代理陷入循环不断重复调用同一工具1. 工具输出未能满足代理的终止条件。2.max_iterations设置过高。3. LLM 对任务理解有误陷入逻辑死循环。1. 查看verbose日志观察每次工具调用的输入输出。2. 检查工具返回的结果格式是否清晰、无歧义。3. 分析代理的“思考”部分看其目标是否在变化。1.优化工具输出确保工具返回明确、结构化的结果例如包含任务完成状态。2.设置迭代限制合理设置max_iterations如5-15。3.改进系统提示词在提示词中明确要求“如果任务已完成或无法继续请直接给出最终答案”。4.引入超时机制在AgentExecutor外层设置任务执行超时。工具调用参数格式错误1. LLM 未能正确理解工具的参数要求。2. 工具函数的文档字符串docstring描述不清。3. 参数类型复杂如嵌套JSON。1. 查看错误日志确认是哪个参数解析失败。2. 对比工具函数签名和代理生成的调用参数。1.完善工具描述在tool装饰器的函数文档中用清晰的语言和示例描述每个参数。2.使用 Pydantic 工具LangChain 支持基于 Pydantic 模型定义工具能提供更严格的类型提示。3.提供少量示例在提示词中加入一两个工具调用的成功示例Few-shot。上下文长度超限1. 对话历史短期记忆过长。2. 从向量数据库检索出的长期记忆片段过多。3. 任务本身过于复杂。1. 监控每次请求的 Token 消耗可通过tiktoken估算。2. 检查ConversationBufferMemory是否积累了过多轮对话。1.使用摘要记忆用ConversationSummaryMemory替代ConversationBufferMemory定期总结历史而非完整保存。2.限制检索数量减少search_memory函数返回的文档数量k值。3.分阶段处理对于超长任务设计外层逻辑将其拆分成多个独立的代理调用。API 调用成本失控1. 代理循环导致过多无效调用。2. 提示词过于冗长。3. 未对用户输入进行过滤。1. 分析日志统计每个用户查询平均消耗的 Token 数和调用次数。2. 检查是否有用户输入触发了异常的复杂推理。1.实施预算与限流在应用层为每个用户/会话设置 Token 消耗上限和调用频率限制。2.优化提示词去除不必要的描述保持简洁。3.输入验证与分类在调用代理前先用一个轻量级模型或规则判断查询意图过滤掉无关或恶意请求。长期记忆检索不准1. 嵌入模型不适合领域文本。2. 存入记忆的文本块chunk过大或过小。3. 检索时相似度阈值设置不当。1. 检查检索返回的内容是否与查询真正相关。2. 分析存入记忆的文本质量。1.领域微调嵌入模型如果条件允许使用领域数据微调嵌入模型。2.优化文本分块根据语义完整性进行分块而不是固定长度。3.重排序Re-ranking在向量检索后使用一个更精细的交叉编码器模型对结果进行重排序提升Top1准确率。5.2 安全与稳定性加固原型可以快速验证想法但生产系统必须考虑安全和稳定。工具执行沙箱化示例中的calculator工具使用了eval这是极其危险的。生产环境中必须使用安全的表达式求值库如ast.literal_eval。或完全实现自己的解析逻辑。或将工具运行在隔离的容器或沙箱环境中。用户输入净化与意图过滤并非所有用户输入都应交给代理处理。应前置一个分类器识别并拦截恶意指令如“删除所有数据”。与代理能力无关的闲聊。试图绕过系统规则的提示词注入Prompt Injection。可观测性与监控全链路日志记录每个用户查询、代理的完整思考链、所有工具调用及结果、最终响应。这对排查问题和优化提示词至关重要。关键指标监控平均响应延迟、Token 消耗分布、工具调用成功率、错误率。审计跟踪对于执行了写操作的工具如发送邮件、更新数据库必须记录“谁在什么时候通过代理执行了什么操作”。优雅降级与超时控制为 LLM API 调用和每个工具调用设置独立的超时。当 LLM 服务或关键工具不可用时应有备选方案如返回缓存结果、提示服务暂时不可用。5.3 性能与成本优化缓存策略LLM 响应缓存对相同的输入提示词缓存 LLM 的输出可以显著降低成本和延迟。可以使用LangChain的LLMCache组件。工具结果缓存对于耗时或费资源的工具调用如复杂查询如果输入参数相同可以缓存其结果。异步执行如果代理需要调用多个独立的 I/O 密集型工具考虑使用异步调用asyncio来并行执行减少总体响应时间。模型选型并非所有任务都需要GPT-4。可以根据任务复杂度设计路由逻辑简单查询用小型/廉价模型复杂规划和创作再用大模型。6. 扩展方向与最佳实践构建出基础代理后你可以根据业务需求向“一人团队”的目标深化。以下是几个关键的扩展方向和实践建议。6.1 扩展方向多代理协作系统模拟一个团队创建不同角色的代理如“规划者”、“执行者”、“审查者”让它们通过共享工作空间或消息队列进行协作处理更复杂的项目。集成真实业务工具将代理与内部系统连接例如数据库工具执行安全的 SQL 查询或更新。API 工具调用内部或第三方 RESTful API。文件操作工具在授权目录下读写文件。通信工具发送邮件、Slack 消息等。强化学习与持续优化记录代理的成功与失败轨迹用于微调底层 LLM 或优化提示词策略使其在特定领域表现更好。本地模型部署出于成本、延迟和数据隐私考虑可以探索使用本地部署的开源模型如 Llama 3、Qwen、DeepSeek通过Ollama、vLLM或LM Studio等框架来驱动代理。这需要解决本地模型的指令跟随和工具调用能力问题。6.2 工程化最佳实践清单在将 AI 代理系统推向生产环境前请对照此清单进行检查[ ]配置外置所有 API Key、模型参数、服务器地址等均通过环境变量或配置中心管理。[ ]依赖管理使用requirements.txt或poetry严格锁定所有依赖包版本。[ ]结构化日志使用structlog或logging模块输出 JSON 格式的日志便于 ELK 等系统收集分析。[ ]异常处理在每个关键步骤LLM调用、工具执行、记忆存储都有明确的异常捕获和恢复机制。[ ]输入验证对用户输入进行长度、字符、敏感词和意图过滤。[ ]输出审查对代理生成的最终输出进行内容安全过滤如防止生成不当内容。[ ]速率限制在应用入口和 LLM API 调用层实施速率限制防止滥用。[ ]版本控制对提示词模板、工具定义、代理配置进行版本控制便于回滚和 A/B 测试。[ ]监控告警设置针对高延迟、高错误率、高 Token 消耗的告警。[ ]数据备份定期备份向量数据库等持久化存储。AI 代理的潜力在于将语言模型的认知能力与外部工具的执行能力无缝结合创造出能真正处理工作流的数字员工。从今天构建的这个原型出发通过持续迭代工具集、优化提示词、加固安全防线并建立完善的监控体系你可以逐步将其打磨成一个能在特定业务场景下可靠运行的“一人团队”。真正的挑战不在于启动第一个循环而在于如何让这个循环在复杂、多变且要求严苛的现实世界中稳定、安全且经济地持续运转下去。