在实际 AI 应用开发中一个核心的挑战是如何让复杂的 AI 能力特别是大语言模型能够稳定、高效地融入现有的工作流和业务系统。开发者常常面临 API 调用不稳定、模型响应格式多变、错误处理复杂、以及如何将 AI 能力编排成自动化流程Agent等问题。近期一个名为 Energy 的开源项目引起了社区的关注它由前 OpenAI 员工推出旨在通过一套简洁的 SDK 和运行时降低 AI 工作流构建的门槛推动 AI 应用的普及。本文将深入探讨如何利用类似 Energy 的设计思想构建一个可复现、可维护的 AI 应用开发框架。我们将从理解 AI Agent 和 Workflow 的核心概念出发逐步完成环境搭建、依赖配置并实现一个具备基础对话、工具调用和状态管理能力的 AI 助手原型。文章的重点不在于复刻某个特定项目而在于掌握构建此类系统的通用工程方法、关键配置参数以及生产环境下的常见问题排查路径。1. 理解 AI Agent 与 Workflow 运行时的核心概念在开始编码之前必须厘清几个关键概念这决定了我们如何设计系统架构。1.1 什么是 AI AgentAI Agent 并非一个全新的概念但在大语言模型时代被赋予了新的内涵。简单来说一个 AI Agent 是一个能够感知环境、进行决策并执行动作以达成目标的系统。在编程语境下它通常指一个封装了 LLM 调用、工具使用如搜索、计算、调用 API和记忆对话历史、知识的软件模块。与直接调用 Chat Completions API 不同一个成熟的 Agent 具备以下特征目标导向它理解用户指令背后的意图并规划步骤去完成。工具使用它可以调用外部函数或 API 来获取信息或执行操作弥补 LLM 在实时性、精确计算等方面的不足。状态管理它能在多轮交互中维持对话历史和任务上下文。自主性与安全性在设定的边界内它可以自主决策下一步行动同时需要有机制防止其执行危险或越权操作。1.2 Workflow 运行时的作用当单个 Agent 无法完成复杂任务时我们需要将多个步骤串联或并联起来这就形成了 Workflow工作流。一个 Workflow 运行时负责编排定义和管理多个 Agent 或工具的执行顺序和依赖关系。状态传递将上一个步骤的输出作为下一个步骤的输入。错误处理与重试当某个步骤失败时决定是重试、跳过还是终止整个流程。并发控制管理并行执行的任务。持久化与可观测性记录工作流的执行日志、中间状态和最终结果便于调试和监控。Energy 这类项目可以看作是一个高度集成化的 Agent Workflow 运行时它提供了标准化的接口和丰富的内置工具让开发者能像搭积木一样构建 AI 应用。1.3 关键组件模型、工具、记忆与提示词构建一个最小可运行的 AI 助手我们需要关注四个核心组件模型Model提供核心推理能力如 OpenAI GPT、 Anthropic Claude 或开源模型。我们需要处理 API 密钥、基础 URL、模型名称等配置。工具Tools扩展 Agent 能力的函数。例如一个获取天气的函数、一个执行数学计算的函数或一个查询数据库的函数。记忆Memory存储和检索对话历史或知识。可以是简单的列表也可以是向量数据库。提示词Prompt定义 Agent 的角色、行为准则和任务指令的模板。良好的提示词工程是 Agent 表现优异的关键。2. 环境准备与项目初始化我们将使用 Python 作为开发语言这是当前 AI 应用开发最活跃的生态。项目将采用清晰的模块化结构。2.1 开发环境与依赖管理首先确保你的开发环境满足以下要求组件要求说明Python3.8 及以上推荐使用 3.10 或 3.11以获得最佳的包兼容性。包管理器pip 或 poetry本文使用pip和venv进行演示。代码编辑器VS Code, PyCharm 等具备良好的 Python 支持即可。接下来创建项目目录并初始化虚拟环境# 创建项目目录 mkdir ai_assistant_framework cd ai_assistant_framework # 创建虚拟环境Windows 用户使用 python -m venv venv python3 -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 升级 pip pip install --upgrade pip2.2 核心依赖安装我们将安装几个核心库来构建我们的框架原型openai用于调用 OpenAI 兼容的 API如 OpenAI 官方、Azure OpenAI 或本地部署的兼容服务。langchain一个流行的框架提供了构建 Agent 和 Chain 的高层抽象。我们将主要使用其工具和记忆组件。pydantic用于数据验证和设置管理确保配置的健壮性。python-dotenv用于从.env文件加载环境变量如 API 密钥。执行以下命令安装pip install openai langchain pydantic python-dotenv注意langchain是一个大型元框架安装时可能会包含许多你暂时用不上的依赖。在生产项目中可以考虑使用langchain-core和langchain-openai等更细粒度的包来减少依赖体积。2.3 项目结构设计一个清晰的项目结构有助于长期维护。创建如下文件和目录ai_assistant_framework/ ├── .env # 存储敏感配置如 API KEY ├── .gitignore # Git 忽略文件 ├── config/ │ └── settings.py # 应用配置管理 ├── core/ │ ├── __init__.py │ ├── agent.py # Agent 核心逻辑 │ ├── tools.py # 自定义工具定义 │ └── memory.py # 记忆管理 ├── models/ │ └── schemas.py # 数据模型定义 ├── main.py # 应用入口 └── requirements.txt # 项目依赖清单生成requirements.txt文件pip freeze requirements.txt在.gitignore文件中至少添加以下内容venv/ .env __pycache__/ *.pyc3. 构建最小化 AI 助手原型现在我们从配置管理开始逐步实现一个能进行对话并使用简单工具的 AI 助手。3.1 配置管理与环境变量首先在项目根目录创建.env文件用于存储敏感信息。切记不要将此文件提交到版本控制系统。# .env OPENAI_API_KEYyour_openai_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用 Azure 或其它兼容服务请修改此处 OPENAI_MODELgpt-3.5-turbo # 或 gpt-4, gpt-4-turbo-preview 等接下来创建config/settings.py使用pydantic来管理和验证配置# config/settings.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): 应用配置类自动从 .env 文件和环境变量加载 openai_api_key: str openai_base_url: Optional[str] https://api.openai.com/v1 openai_model: str gpt-3.5-turbo # 可以添加其他配置如日志级别、数据库连接等 # log_level: str INFO class Config: env_file .env env_file_encoding utf-8 # 创建全局配置实例 settings Settings()注意这里使用了pydantic-settings它是pydantic的扩展专门用于设置管理。如果未安装请运行pip install pydantic-settings。使用BaseSettings可以确保配置缺失时得到清晰的错误提示而不是在运行时因None值而崩溃。3.2 定义自定义工具工具是 Agent 的手臂。我们在core/tools.py中定义两个简单的工具一个计算器和一个获取当前时间的工具。# core/tools.py from datetime import datetime from typing import Type from pydantic import BaseModel, Field from langchain.tools import BaseTool class CalculatorInput(BaseModel): 计算器工具的输入模型 expression: str Field(description一个合法的数学表达式例如3 5 * 2) class CalculatorTool(BaseTool): 一个简单的数学计算器工具 name calculator description 用于计算一个数学表达式的值。输入应该是一个字符串格式的表达式。 args_schema: Type[BaseModel] CalculatorInput def _run(self, expression: str) - str: 执行计算 try: # 警告使用 eval 存在安全风险仅用于演示。 # 生产环境应使用更安全的表达式求值库如 ast.literal_eval 或自定义解析器。 result eval(expression, {__builtins__: {}}, {}) return f表达式 {expression} 的计算结果是{result} except Exception as e: return f计算失败{e} async def _arun(self, expression: str) - str: 异步执行本例中与同步相同 return self._run(expression) def get_current_time(*args, **kwargs) - str: 获取当前日期和时间的工具函数 now datetime.now() return f当前时间是{now.strftime(%Y-%m-%d %H:%M:%S)} # 工具集合 def get_all_tools(): 返回所有可用工具的列表 return [ CalculatorTool(), # 使用 Tool.from_function 包装普通函数 Tool.from_function( funcget_current_time, nameget_current_time, description获取当前的日期和时间。 ) ]关键点解释输入验证使用pydantic的BaseModel定义工具输入的结构这能帮助 LLM 生成正确的调用参数。描述清晰name和description必须准确因为 LLM 依赖这些描述来决定何时以及如何使用工具。安全警告示例中的eval函数极不安全绝对不能用于处理用户不可信的输入。实际项目中必须替换为安全的计算引擎。3.3 实现带有记忆的 Agent 核心在core/agent.py中我们将组合模型、工具和记忆构建 Agent 的核心逻辑。# core/agent.py from typing import List, Any from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain.memory import ConversationBufferMemory from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from config.settings import settings from core.tools import get_all_tools class AssistantAgent: AI 助手 Agent 类 def __init__(self): # 1. 初始化 LLM self.llm ChatOpenAI( modelsettings.openai_model, openai_api_keysettings.openai_api_key, base_urlsettings.openai_base_url, temperature0.1, # 较低的温度使输出更稳定、更可预测 ) # 2. 获取工具列表 self.tools get_all_tools() # 3. 构建提示词模板 # 提示词定义了 Agent 的角色和行为指令 self.prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的 AI 助手。你可以使用工具来回答问题。 如果你不知道答案就诚实地回答不知道不要编造信息。 使用工具时请清晰地说明你正在做什么。), MessagesPlaceholder(variable_namechat_history), # 历史消息占位符 (human, {input}), # 用户当前输入占位符 MessagesPlaceholder(variable_nameagent_scratchpad), # Agent 思考过程占位符 ]) # 4. 初始化记忆存储对话历史 self.memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue # 返回 Message 对象列表而非字符串 ) # 5. 创建 Agent agent create_openai_tools_agent( llmself.llm, toolsself.tools, promptself.prompt ) # 6. 创建 Agent 执行器它负责调用 Agent 并管理工具执行循环 self.agent_executor AgentExecutor( agentagent, toolsself.tools, memoryself.memory, verboseTrue, # 设置为 True 可以在控制台看到详细的思考过程 handle_parsing_errorsTrue, # 处理 Agent 输出解析错误 max_iterations5, # 限制最大迭代次数防止无限循环 ) def run(self, user_input: str) - str: 运行 Agent处理用户输入 try: response self.agent_executor.invoke({input: user_input}) return response[output] except Exception as e: # 处理执行过程中的异常 return f抱歉处理您的请求时出现了错误{str(e)} # 创建全局 Agent 实例单例模式简单演示 agent_instance AssistantAgent()代码详解ChatOpenAILangChain 对 OpenAI API 的封装。通过base_url参数可以轻松切换到 Azure OpenAI 或其他兼容 OpenAI API 的服务如 DashScope 的兼容地址。create_openai_tools_agent这是 LangChain 提供的一个高级函数它基于 OpenAI 的function calling能力自动构建一个能理解和使用工具的 Agent。ConversationBufferMemory一个简单的内存实现将整个对话历史保存在内存中。对于长对话可以考虑使用ConversationSummaryMemory或向量数据库。AgentExecutor这是真正的“运行时”。它接收用户输入和记忆驱动 Agent 进行“思考 - 决定是否调用工具 - 获取工具结果 - 继续思考”的循环直到 Agent 决定给出最终答案或达到max_iterations限制。verboseTrue这是一个非常重要的调试选项。当设置为True时控制台会打印出 Agent 的完整思考链Chain of Thought包括它决定调用哪个工具、工具返回的结果等这对于排查 Agent 行为异常至关重要。3.4 创建应用入口并运行测试最后在main.py中创建一个简单的交互循环来测试我们的 Agent。# main.py import sys from core.agent import agent_instance def main(): print(AI 助手已启动。输入 退出 或 quit 来结束对话。) print(- * 40) while True: try: user_input input(\n你: ).strip() except (EOFError, KeyboardInterrupt): print(\n再见) sys.exit(0) if user_input.lower() in [退出, quit, exit]: print(助手: 再见) break if not user_input: continue # 调用 Agent 获取回复 print(助手: 思考中...) response agent_instance.run(user_input) print(f助手: {response}) if __name__ __main__: main()现在运行你的 AI 助手# 确保在项目根目录且虚拟环境已激活 python main.py你应该能看到类似以下的交互过程verboseTrue时的输出AI 助手已启动。输入 退出 或 quit 来结束对话。 ---------------------------------------- 你: 3 加 5 等于多少 助手: 思考中... 进入新的 Agent 执行链... 我可以用计算器工具来计算 35。 Action: calculator Action Input: {expression: 3 5} Observation: 表达式 3 5 的计算结果是8 Thought: 用户问 3 加 5 等于多少我通过计算器工具得到结果是 8。 Action: Action Input: {} Observation: Thought: 我已经得到了答案可以回复用户了。 助手: 3 加 5 等于 8。 你: 现在几点了 ...4. 关键配置、参数详解与生产考量我们的原型已经可以运行但要将其用于更严肃的场景必须理解并调整关键参数。4.1 模型与 API 配置参数在config/settings.py中我们定义了基础配置。以下是关键参数的解释和调优建议参数类型默认值/示例说明与影响openai_api_keystrsk-...必填。身份凭证。生产环境务必通过环境变量或密钥管理服务注入切勿硬编码。openai_base_urlOptional[str]https://api.openai.com/v1API 端点。如果使用 Azure OpenAI格式为https://{resource}.openai.azure.com/openai/deployments/{deployment}。使用第三方兼容服务时修改此值。openai_modelstrgpt-3.5-turbo模型名称。对于 Azure填写部署名称。不同模型在成本、速度、上下文长度和能力上有差异。temperaturefloat0.1(在代码中设置)采样温度范围 0~2。值越低输出越确定、重复值越高输出越随机、有创造性。对于工具调用等任务建议较低值如 0.1-0.3。max_tokensintNone(默认)生成的最大 token 数。设置上限可控制成本但可能导致回答被截断。request_timeoutfloatNone(默认)请求超时时间秒。生产环境必须设置如 30.0避免线程阻塞。4.2 Agent 执行器控制参数在AgentExecutor初始化时有几个参数控制着执行行为和安全边界参数说明与生产建议verbose调试时设为True生产环境设为False以减少日志噪音。可通过环境变量动态控制。handle_parsing_errors必须设为True。当 Agent 输出无法解析为工具调用或最终答案时此参数决定是否进行错误恢复例如让 LLM 重试。max_iterations至关重要。限制 Agent 单次调用的最大“思考-行动”循环次数防止因逻辑错误或工具失败导致无限循环和 API 费用激增。根据任务复杂度设置通常 5-10 次足够。early_stopping_method提前停止方法。例如“force”会在达到max_iterations时强制返回当前结果。return_intermediate_steps设为True可在返回结果中包含中间步骤工具调用记录用于审计或复杂工作流的后续处理。4.3 记忆Memory的选型与配置我们使用了简单的ConversationBufferMemory它将所有对话历史以明文存储在内存中。在生产环境中你需要根据场景选择记忆类型适用场景注意事项ConversationBufferMemory对话轮次少、上下文短的场景。历史越长消耗的 Token 越多成本越高且可能超出模型上下文窗口。ConversationSummaryMemory长对话场景。LLM 会定期总结历史将总结而非原文放入上下文。节省 Token但可能丢失细节。需要额外的 LLM 调用成本。ConversationBufferWindowMemory需要最近 N 轮对话的场景。只保留最近 K 轮对话是缓冲和窗口的折中方案。VectorStoreRetrieverMemory需要从大量历史或知识库中检索相关片段的场景。结合向量数据库根据当前查询语义检索最相关的历史片段。架构复杂但能力最强。生产建议对于客服机器人等长对话应用优先考虑ConversationSummaryMemory或ConversationBufferWindowMemory。对于需要长期、精准记忆的智能体需设计基于向量数据库的外挂记忆系统。5. 常见问题排查与调试指南在开发和运行 AI Agent 时你会遇到一些典型问题。以下是系统的排查路径。5.1 Agent 不调用工具总是直接回答现象即使问题明显需要计算或查询Agent 也尝试用模型自身的知识直接回答而不触发工具调用。可能原因与排查工具描述不清晰检查core/tools.py中每个工具的name和description是否准确描述了工具的功能和适用场景。LLM 完全依赖这些描述做决策。提示词未引导检查core/agent.py中的系统提示词systemmessage。是否明确鼓励或指示 Agent 使用工具可以加入“请优先使用工具来获取准确信息”等指令。模型能力不足较旧的或能力较弱的模型如gpt-3.5-turbo的某些版本的函数调用/工具使用能力可能不稳定。尝试切换到更新的模型如gpt-3.5-turbo-0125或gpt-4。温度Temperature过高过高的temperature可能导致输出随机性太大无法稳定触发工具调用逻辑。尝试将其降低到 0.1 或 0.2。5.2 遇到 “InvalidRequestError” 或 “AuthenticationError”现象控制台抛出与 API 请求相关的错误。排查步骤检查 API 密钥确认.env文件中的OPENAI_API_KEY正确无误且没有多余空格。可以通过在 Python 交互环境中直接使用openai库发起一个简单请求来验证。import openai openai.api_key your_key # 测试调用 try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[{role: user, content: Hello}], max_tokens5 ) print(API 密钥有效) except Exception as e: print(fAPI 调用失败: {e})检查 Base URL如果使用非官方 OpenAI 服务确认OPENAI_BASE_URL完全正确并且该服务支持 Chat Completions 和 Function Calling 接口。检查网络与代理确保你的网络环境可以访问目标 API 地址。公司网络或某些地区可能需要配置网络策略。检查配额与账单登录 OpenAI 或相应云服务平台确认账户是否有可用额度是否欠费。5.3 Agent 陷入无限循环或达到最大迭代次数现象Agent 反复调用同一个工具或者在不同工具间来回切换始终无法给出最终答案最终因max_iterations限制而停止。排查与解决开启详细日志确保AgentExecutor(verboseTrue)。观察控制台输出的Thought、Action和Observation。这能清晰展示 Agent 的思考过程。分析工具观察结果检查工具返回给 Agent 的Observation是否清晰、格式正确。模糊或错误的观察结果会导致 Agent 误解。优化工具设计工具应具有原子性一个工具只做一件事并明确返回成功或失败。工具应能处理边界情况例如计算器工具在遇到除零错误时应返回明确的错误信息而不是抛出异常导致整个流程中断。避免工具链循环依赖确保工具 A 不会在某种条件下总是调用工具 B而工具 B 又反过来调用工具 A。调整提示词在系统提示词中增加约束例如“如果尝试了两次仍无法解决问题请总结当前已知信息并告知用户而不是继续尝试”。5.4 处理解析错误Parsing Errors现象控制台出现OutputParserException等解析错误。原因Agent 的输出不符合 LangChain 预期的格式例如工具调用的 JSON 格式错误。解决方案确保AgentExecutor初始化时handle_parsing_errorsTrue。这会让执行器尝试从错误中恢复。如果问题持续检查是否因模型输出不稳定导致。可以尝试降低temperature或使用更强大的模型。在极少数情况下可能是 LangChain 的解析器与模型输出不兼容。可以查阅对应版本的 LangChain 文档。6. 从原型到生产最佳实践与扩展方向一个可学习的原型与一个健壮的生产系统之间存在巨大差距。以下是关键的进阶步骤。6.1 安全与权限控制工具沙箱化绝对不要像示例中那样使用eval。任何执行代码、访问文件系统或网络资源的工具都必须运行在严格的沙箱环境中并对输入进行白名单验证。用户输入净化对所有传入 Agent 的用户输入进行必要的清洗和检查防止提示词注入攻击。工具访问权限为不同的用户或会话定义工具白名单。不是所有用户都能使用所有工具。内容安全过滤在 Agent 的最终输出返回给用户前加入一层内容安全过滤防止生成有害或不适当的内容。6.2 可观测性与监控结构化日志替换print语句使用logging模块记录结构化日志。记录每次调用的用户 ID、会话 ID、输入、输出、使用的工具、消耗的 Token 数和耗时。链路追踪为每个用户请求生成唯一的trace_id并在所有后续的模型调用、工具调用中传递它。这能让你在分布式系统中完整追踪一个请求的路径。关键指标监控延迟请求响应时间 P95/P99。成本每日/每用户的 Token 消耗和 API 调用费用。错误率工具调用失败率、模型调用错误率。有效性用户反馈如点赞/点踩或业务成功率。6.3 性能与成本优化缓存对频繁出现的、结果确定的用户查询如“公司的退货政策是什么”的 LLM 响应进行缓存。可以使用 Redis 或内存缓存。异步处理对于耗时较长的工具调用如调用一个慢速的外部 API使用异步 Agent (AgentExecutor.ainvoke)避免阻塞主线程。模型选型根据任务难度选择合适的模型。简单的分类、提取任务可以使用小模型复杂的推理、规划任务再使用大模型。实现一个路由层根据意图判断分派给不同模型。上下文管理积极管理对话上下文。定期总结或清除陈旧的历史避免不必要的 Token 消耗。6.4 架构扩展实现 Workflow当单个 Agent 不够时你需要 Workflow 运行时。这通常意味着定义工作流 DSL使用 JSON、YAML 或 Python 代码来定义一组任务及其依赖关系。实现任务节点每个节点可以是一个 Agent、一个工具或一个条件判断。添加控制流支持顺序、并行、分支if-else、循环for/while等逻辑。状态持久化将工作流执行状态保存到数据库中支持暂停、恢复和重试。实现调度器一个中心化的组件负责解析工作流定义、调度任务执行、处理错误和重试。你可以从实现一个简单的顺序执行器开始逐步增加复杂度。也可以评估现有的开源 Workflow 引擎如 Prefect、Airflow是否适合与你的 AI Agent 集成。构建一个成熟的 AI Agent 系统是一个持续迭代的过程。从本文的最小原型出发围绕具体的业务需求在安全性、可靠性、可观测性和成本控制四个维度上不断深化才能真正让 AI 能力可靠地普及到各类工作流中。下一步你可以尝试集成一个真实的 API如天气查询或者为 Agent 添加从向量数据库检索知识的能力。
从零构建AI Agent框架：基于LangChain的工程实践与生产部署指南
在实际 AI 应用开发中一个核心的挑战是如何让复杂的 AI 能力特别是大语言模型能够稳定、高效地融入现有的工作流和业务系统。开发者常常面临 API 调用不稳定、模型响应格式多变、错误处理复杂、以及如何将 AI 能力编排成自动化流程Agent等问题。近期一个名为 Energy 的开源项目引起了社区的关注它由前 OpenAI 员工推出旨在通过一套简洁的 SDK 和运行时降低 AI 工作流构建的门槛推动 AI 应用的普及。本文将深入探讨如何利用类似 Energy 的设计思想构建一个可复现、可维护的 AI 应用开发框架。我们将从理解 AI Agent 和 Workflow 的核心概念出发逐步完成环境搭建、依赖配置并实现一个具备基础对话、工具调用和状态管理能力的 AI 助手原型。文章的重点不在于复刻某个特定项目而在于掌握构建此类系统的通用工程方法、关键配置参数以及生产环境下的常见问题排查路径。1. 理解 AI Agent 与 Workflow 运行时的核心概念在开始编码之前必须厘清几个关键概念这决定了我们如何设计系统架构。1.1 什么是 AI AgentAI Agent 并非一个全新的概念但在大语言模型时代被赋予了新的内涵。简单来说一个 AI Agent 是一个能够感知环境、进行决策并执行动作以达成目标的系统。在编程语境下它通常指一个封装了 LLM 调用、工具使用如搜索、计算、调用 API和记忆对话历史、知识的软件模块。与直接调用 Chat Completions API 不同一个成熟的 Agent 具备以下特征目标导向它理解用户指令背后的意图并规划步骤去完成。工具使用它可以调用外部函数或 API 来获取信息或执行操作弥补 LLM 在实时性、精确计算等方面的不足。状态管理它能在多轮交互中维持对话历史和任务上下文。自主性与安全性在设定的边界内它可以自主决策下一步行动同时需要有机制防止其执行危险或越权操作。1.2 Workflow 运行时的作用当单个 Agent 无法完成复杂任务时我们需要将多个步骤串联或并联起来这就形成了 Workflow工作流。一个 Workflow 运行时负责编排定义和管理多个 Agent 或工具的执行顺序和依赖关系。状态传递将上一个步骤的输出作为下一个步骤的输入。错误处理与重试当某个步骤失败时决定是重试、跳过还是终止整个流程。并发控制管理并行执行的任务。持久化与可观测性记录工作流的执行日志、中间状态和最终结果便于调试和监控。Energy 这类项目可以看作是一个高度集成化的 Agent Workflow 运行时它提供了标准化的接口和丰富的内置工具让开发者能像搭积木一样构建 AI 应用。1.3 关键组件模型、工具、记忆与提示词构建一个最小可运行的 AI 助手我们需要关注四个核心组件模型Model提供核心推理能力如 OpenAI GPT、 Anthropic Claude 或开源模型。我们需要处理 API 密钥、基础 URL、模型名称等配置。工具Tools扩展 Agent 能力的函数。例如一个获取天气的函数、一个执行数学计算的函数或一个查询数据库的函数。记忆Memory存储和检索对话历史或知识。可以是简单的列表也可以是向量数据库。提示词Prompt定义 Agent 的角色、行为准则和任务指令的模板。良好的提示词工程是 Agent 表现优异的关键。2. 环境准备与项目初始化我们将使用 Python 作为开发语言这是当前 AI 应用开发最活跃的生态。项目将采用清晰的模块化结构。2.1 开发环境与依赖管理首先确保你的开发环境满足以下要求组件要求说明Python3.8 及以上推荐使用 3.10 或 3.11以获得最佳的包兼容性。包管理器pip 或 poetry本文使用pip和venv进行演示。代码编辑器VS Code, PyCharm 等具备良好的 Python 支持即可。接下来创建项目目录并初始化虚拟环境# 创建项目目录 mkdir ai_assistant_framework cd ai_assistant_framework # 创建虚拟环境Windows 用户使用 python -m venv venv python3 -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows: # venv\Scripts\activate # 升级 pip pip install --upgrade pip2.2 核心依赖安装我们将安装几个核心库来构建我们的框架原型openai用于调用 OpenAI 兼容的 API如 OpenAI 官方、Azure OpenAI 或本地部署的兼容服务。langchain一个流行的框架提供了构建 Agent 和 Chain 的高层抽象。我们将主要使用其工具和记忆组件。pydantic用于数据验证和设置管理确保配置的健壮性。python-dotenv用于从.env文件加载环境变量如 API 密钥。执行以下命令安装pip install openai langchain pydantic python-dotenv注意langchain是一个大型元框架安装时可能会包含许多你暂时用不上的依赖。在生产项目中可以考虑使用langchain-core和langchain-openai等更细粒度的包来减少依赖体积。2.3 项目结构设计一个清晰的项目结构有助于长期维护。创建如下文件和目录ai_assistant_framework/ ├── .env # 存储敏感配置如 API KEY ├── .gitignore # Git 忽略文件 ├── config/ │ └── settings.py # 应用配置管理 ├── core/ │ ├── __init__.py │ ├── agent.py # Agent 核心逻辑 │ ├── tools.py # 自定义工具定义 │ └── memory.py # 记忆管理 ├── models/ │ └── schemas.py # 数据模型定义 ├── main.py # 应用入口 └── requirements.txt # 项目依赖清单生成requirements.txt文件pip freeze requirements.txt在.gitignore文件中至少添加以下内容venv/ .env __pycache__/ *.pyc3. 构建最小化 AI 助手原型现在我们从配置管理开始逐步实现一个能进行对话并使用简单工具的 AI 助手。3.1 配置管理与环境变量首先在项目根目录创建.env文件用于存储敏感信息。切记不要将此文件提交到版本控制系统。# .env OPENAI_API_KEYyour_openai_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 # 如果使用 Azure 或其它兼容服务请修改此处 OPENAI_MODELgpt-3.5-turbo # 或 gpt-4, gpt-4-turbo-preview 等接下来创建config/settings.py使用pydantic来管理和验证配置# config/settings.py from pydantic_settings import BaseSettings from typing import Optional class Settings(BaseSettings): 应用配置类自动从 .env 文件和环境变量加载 openai_api_key: str openai_base_url: Optional[str] https://api.openai.com/v1 openai_model: str gpt-3.5-turbo # 可以添加其他配置如日志级别、数据库连接等 # log_level: str INFO class Config: env_file .env env_file_encoding utf-8 # 创建全局配置实例 settings Settings()注意这里使用了pydantic-settings它是pydantic的扩展专门用于设置管理。如果未安装请运行pip install pydantic-settings。使用BaseSettings可以确保配置缺失时得到清晰的错误提示而不是在运行时因None值而崩溃。3.2 定义自定义工具工具是 Agent 的手臂。我们在core/tools.py中定义两个简单的工具一个计算器和一个获取当前时间的工具。# core/tools.py from datetime import datetime from typing import Type from pydantic import BaseModel, Field from langchain.tools import BaseTool class CalculatorInput(BaseModel): 计算器工具的输入模型 expression: str Field(description一个合法的数学表达式例如3 5 * 2) class CalculatorTool(BaseTool): 一个简单的数学计算器工具 name calculator description 用于计算一个数学表达式的值。输入应该是一个字符串格式的表达式。 args_schema: Type[BaseModel] CalculatorInput def _run(self, expression: str) - str: 执行计算 try: # 警告使用 eval 存在安全风险仅用于演示。 # 生产环境应使用更安全的表达式求值库如 ast.literal_eval 或自定义解析器。 result eval(expression, {__builtins__: {}}, {}) return f表达式 {expression} 的计算结果是{result} except Exception as e: return f计算失败{e} async def _arun(self, expression: str) - str: 异步执行本例中与同步相同 return self._run(expression) def get_current_time(*args, **kwargs) - str: 获取当前日期和时间的工具函数 now datetime.now() return f当前时间是{now.strftime(%Y-%m-%d %H:%M:%S)} # 工具集合 def get_all_tools(): 返回所有可用工具的列表 return [ CalculatorTool(), # 使用 Tool.from_function 包装普通函数 Tool.from_function( funcget_current_time, nameget_current_time, description获取当前的日期和时间。 ) ]关键点解释输入验证使用pydantic的BaseModel定义工具输入的结构这能帮助 LLM 生成正确的调用参数。描述清晰name和description必须准确因为 LLM 依赖这些描述来决定何时以及如何使用工具。安全警告示例中的eval函数极不安全绝对不能用于处理用户不可信的输入。实际项目中必须替换为安全的计算引擎。3.3 实现带有记忆的 Agent 核心在core/agent.py中我们将组合模型、工具和记忆构建 Agent 的核心逻辑。# core/agent.py from typing import List, Any from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain.memory import ConversationBufferMemory from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from config.settings import settings from core.tools import get_all_tools class AssistantAgent: AI 助手 Agent 类 def __init__(self): # 1. 初始化 LLM self.llm ChatOpenAI( modelsettings.openai_model, openai_api_keysettings.openai_api_key, base_urlsettings.openai_base_url, temperature0.1, # 较低的温度使输出更稳定、更可预测 ) # 2. 获取工具列表 self.tools get_all_tools() # 3. 构建提示词模板 # 提示词定义了 Agent 的角色和行为指令 self.prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的 AI 助手。你可以使用工具来回答问题。 如果你不知道答案就诚实地回答不知道不要编造信息。 使用工具时请清晰地说明你正在做什么。), MessagesPlaceholder(variable_namechat_history), # 历史消息占位符 (human, {input}), # 用户当前输入占位符 MessagesPlaceholder(variable_nameagent_scratchpad), # Agent 思考过程占位符 ]) # 4. 初始化记忆存储对话历史 self.memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue # 返回 Message 对象列表而非字符串 ) # 5. 创建 Agent agent create_openai_tools_agent( llmself.llm, toolsself.tools, promptself.prompt ) # 6. 创建 Agent 执行器它负责调用 Agent 并管理工具执行循环 self.agent_executor AgentExecutor( agentagent, toolsself.tools, memoryself.memory, verboseTrue, # 设置为 True 可以在控制台看到详细的思考过程 handle_parsing_errorsTrue, # 处理 Agent 输出解析错误 max_iterations5, # 限制最大迭代次数防止无限循环 ) def run(self, user_input: str) - str: 运行 Agent处理用户输入 try: response self.agent_executor.invoke({input: user_input}) return response[output] except Exception as e: # 处理执行过程中的异常 return f抱歉处理您的请求时出现了错误{str(e)} # 创建全局 Agent 实例单例模式简单演示 agent_instance AssistantAgent()代码详解ChatOpenAILangChain 对 OpenAI API 的封装。通过base_url参数可以轻松切换到 Azure OpenAI 或其他兼容 OpenAI API 的服务如 DashScope 的兼容地址。create_openai_tools_agent这是 LangChain 提供的一个高级函数它基于 OpenAI 的function calling能力自动构建一个能理解和使用工具的 Agent。ConversationBufferMemory一个简单的内存实现将整个对话历史保存在内存中。对于长对话可以考虑使用ConversationSummaryMemory或向量数据库。AgentExecutor这是真正的“运行时”。它接收用户输入和记忆驱动 Agent 进行“思考 - 决定是否调用工具 - 获取工具结果 - 继续思考”的循环直到 Agent 决定给出最终答案或达到max_iterations限制。verboseTrue这是一个非常重要的调试选项。当设置为True时控制台会打印出 Agent 的完整思考链Chain of Thought包括它决定调用哪个工具、工具返回的结果等这对于排查 Agent 行为异常至关重要。3.4 创建应用入口并运行测试最后在main.py中创建一个简单的交互循环来测试我们的 Agent。# main.py import sys from core.agent import agent_instance def main(): print(AI 助手已启动。输入 退出 或 quit 来结束对话。) print(- * 40) while True: try: user_input input(\n你: ).strip() except (EOFError, KeyboardInterrupt): print(\n再见) sys.exit(0) if user_input.lower() in [退出, quit, exit]: print(助手: 再见) break if not user_input: continue # 调用 Agent 获取回复 print(助手: 思考中...) response agent_instance.run(user_input) print(f助手: {response}) if __name__ __main__: main()现在运行你的 AI 助手# 确保在项目根目录且虚拟环境已激活 python main.py你应该能看到类似以下的交互过程verboseTrue时的输出AI 助手已启动。输入 退出 或 quit 来结束对话。 ---------------------------------------- 你: 3 加 5 等于多少 助手: 思考中... 进入新的 Agent 执行链... 我可以用计算器工具来计算 35。 Action: calculator Action Input: {expression: 3 5} Observation: 表达式 3 5 的计算结果是8 Thought: 用户问 3 加 5 等于多少我通过计算器工具得到结果是 8。 Action: Action Input: {} Observation: Thought: 我已经得到了答案可以回复用户了。 助手: 3 加 5 等于 8。 你: 现在几点了 ...4. 关键配置、参数详解与生产考量我们的原型已经可以运行但要将其用于更严肃的场景必须理解并调整关键参数。4.1 模型与 API 配置参数在config/settings.py中我们定义了基础配置。以下是关键参数的解释和调优建议参数类型默认值/示例说明与影响openai_api_keystrsk-...必填。身份凭证。生产环境务必通过环境变量或密钥管理服务注入切勿硬编码。openai_base_urlOptional[str]https://api.openai.com/v1API 端点。如果使用 Azure OpenAI格式为https://{resource}.openai.azure.com/openai/deployments/{deployment}。使用第三方兼容服务时修改此值。openai_modelstrgpt-3.5-turbo模型名称。对于 Azure填写部署名称。不同模型在成本、速度、上下文长度和能力上有差异。temperaturefloat0.1(在代码中设置)采样温度范围 0~2。值越低输出越确定、重复值越高输出越随机、有创造性。对于工具调用等任务建议较低值如 0.1-0.3。max_tokensintNone(默认)生成的最大 token 数。设置上限可控制成本但可能导致回答被截断。request_timeoutfloatNone(默认)请求超时时间秒。生产环境必须设置如 30.0避免线程阻塞。4.2 Agent 执行器控制参数在AgentExecutor初始化时有几个参数控制着执行行为和安全边界参数说明与生产建议verbose调试时设为True生产环境设为False以减少日志噪音。可通过环境变量动态控制。handle_parsing_errors必须设为True。当 Agent 输出无法解析为工具调用或最终答案时此参数决定是否进行错误恢复例如让 LLM 重试。max_iterations至关重要。限制 Agent 单次调用的最大“思考-行动”循环次数防止因逻辑错误或工具失败导致无限循环和 API 费用激增。根据任务复杂度设置通常 5-10 次足够。early_stopping_method提前停止方法。例如“force”会在达到max_iterations时强制返回当前结果。return_intermediate_steps设为True可在返回结果中包含中间步骤工具调用记录用于审计或复杂工作流的后续处理。4.3 记忆Memory的选型与配置我们使用了简单的ConversationBufferMemory它将所有对话历史以明文存储在内存中。在生产环境中你需要根据场景选择记忆类型适用场景注意事项ConversationBufferMemory对话轮次少、上下文短的场景。历史越长消耗的 Token 越多成本越高且可能超出模型上下文窗口。ConversationSummaryMemory长对话场景。LLM 会定期总结历史将总结而非原文放入上下文。节省 Token但可能丢失细节。需要额外的 LLM 调用成本。ConversationBufferWindowMemory需要最近 N 轮对话的场景。只保留最近 K 轮对话是缓冲和窗口的折中方案。VectorStoreRetrieverMemory需要从大量历史或知识库中检索相关片段的场景。结合向量数据库根据当前查询语义检索最相关的历史片段。架构复杂但能力最强。生产建议对于客服机器人等长对话应用优先考虑ConversationSummaryMemory或ConversationBufferWindowMemory。对于需要长期、精准记忆的智能体需设计基于向量数据库的外挂记忆系统。5. 常见问题排查与调试指南在开发和运行 AI Agent 时你会遇到一些典型问题。以下是系统的排查路径。5.1 Agent 不调用工具总是直接回答现象即使问题明显需要计算或查询Agent 也尝试用模型自身的知识直接回答而不触发工具调用。可能原因与排查工具描述不清晰检查core/tools.py中每个工具的name和description是否准确描述了工具的功能和适用场景。LLM 完全依赖这些描述做决策。提示词未引导检查core/agent.py中的系统提示词systemmessage。是否明确鼓励或指示 Agent 使用工具可以加入“请优先使用工具来获取准确信息”等指令。模型能力不足较旧的或能力较弱的模型如gpt-3.5-turbo的某些版本的函数调用/工具使用能力可能不稳定。尝试切换到更新的模型如gpt-3.5-turbo-0125或gpt-4。温度Temperature过高过高的temperature可能导致输出随机性太大无法稳定触发工具调用逻辑。尝试将其降低到 0.1 或 0.2。5.2 遇到 “InvalidRequestError” 或 “AuthenticationError”现象控制台抛出与 API 请求相关的错误。排查步骤检查 API 密钥确认.env文件中的OPENAI_API_KEY正确无误且没有多余空格。可以通过在 Python 交互环境中直接使用openai库发起一个简单请求来验证。import openai openai.api_key your_key # 测试调用 try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[{role: user, content: Hello}], max_tokens5 ) print(API 密钥有效) except Exception as e: print(fAPI 调用失败: {e})检查 Base URL如果使用非官方 OpenAI 服务确认OPENAI_BASE_URL完全正确并且该服务支持 Chat Completions 和 Function Calling 接口。检查网络与代理确保你的网络环境可以访问目标 API 地址。公司网络或某些地区可能需要配置网络策略。检查配额与账单登录 OpenAI 或相应云服务平台确认账户是否有可用额度是否欠费。5.3 Agent 陷入无限循环或达到最大迭代次数现象Agent 反复调用同一个工具或者在不同工具间来回切换始终无法给出最终答案最终因max_iterations限制而停止。排查与解决开启详细日志确保AgentExecutor(verboseTrue)。观察控制台输出的Thought、Action和Observation。这能清晰展示 Agent 的思考过程。分析工具观察结果检查工具返回给 Agent 的Observation是否清晰、格式正确。模糊或错误的观察结果会导致 Agent 误解。优化工具设计工具应具有原子性一个工具只做一件事并明确返回成功或失败。工具应能处理边界情况例如计算器工具在遇到除零错误时应返回明确的错误信息而不是抛出异常导致整个流程中断。避免工具链循环依赖确保工具 A 不会在某种条件下总是调用工具 B而工具 B 又反过来调用工具 A。调整提示词在系统提示词中增加约束例如“如果尝试了两次仍无法解决问题请总结当前已知信息并告知用户而不是继续尝试”。5.4 处理解析错误Parsing Errors现象控制台出现OutputParserException等解析错误。原因Agent 的输出不符合 LangChain 预期的格式例如工具调用的 JSON 格式错误。解决方案确保AgentExecutor初始化时handle_parsing_errorsTrue。这会让执行器尝试从错误中恢复。如果问题持续检查是否因模型输出不稳定导致。可以尝试降低temperature或使用更强大的模型。在极少数情况下可能是 LangChain 的解析器与模型输出不兼容。可以查阅对应版本的 LangChain 文档。6. 从原型到生产最佳实践与扩展方向一个可学习的原型与一个健壮的生产系统之间存在巨大差距。以下是关键的进阶步骤。6.1 安全与权限控制工具沙箱化绝对不要像示例中那样使用eval。任何执行代码、访问文件系统或网络资源的工具都必须运行在严格的沙箱环境中并对输入进行白名单验证。用户输入净化对所有传入 Agent 的用户输入进行必要的清洗和检查防止提示词注入攻击。工具访问权限为不同的用户或会话定义工具白名单。不是所有用户都能使用所有工具。内容安全过滤在 Agent 的最终输出返回给用户前加入一层内容安全过滤防止生成有害或不适当的内容。6.2 可观测性与监控结构化日志替换print语句使用logging模块记录结构化日志。记录每次调用的用户 ID、会话 ID、输入、输出、使用的工具、消耗的 Token 数和耗时。链路追踪为每个用户请求生成唯一的trace_id并在所有后续的模型调用、工具调用中传递它。这能让你在分布式系统中完整追踪一个请求的路径。关键指标监控延迟请求响应时间 P95/P99。成本每日/每用户的 Token 消耗和 API 调用费用。错误率工具调用失败率、模型调用错误率。有效性用户反馈如点赞/点踩或业务成功率。6.3 性能与成本优化缓存对频繁出现的、结果确定的用户查询如“公司的退货政策是什么”的 LLM 响应进行缓存。可以使用 Redis 或内存缓存。异步处理对于耗时较长的工具调用如调用一个慢速的外部 API使用异步 Agent (AgentExecutor.ainvoke)避免阻塞主线程。模型选型根据任务难度选择合适的模型。简单的分类、提取任务可以使用小模型复杂的推理、规划任务再使用大模型。实现一个路由层根据意图判断分派给不同模型。上下文管理积极管理对话上下文。定期总结或清除陈旧的历史避免不必要的 Token 消耗。6.4 架构扩展实现 Workflow当单个 Agent 不够时你需要 Workflow 运行时。这通常意味着定义工作流 DSL使用 JSON、YAML 或 Python 代码来定义一组任务及其依赖关系。实现任务节点每个节点可以是一个 Agent、一个工具或一个条件判断。添加控制流支持顺序、并行、分支if-else、循环for/while等逻辑。状态持久化将工作流执行状态保存到数据库中支持暂停、恢复和重试。实现调度器一个中心化的组件负责解析工作流定义、调度任务执行、处理错误和重试。你可以从实现一个简单的顺序执行器开始逐步增加复杂度。也可以评估现有的开源 Workflow 引擎如 Prefect、Airflow是否适合与你的 AI Agent 集成。构建一个成熟的 AI Agent 系统是一个持续迭代的过程。从本文的最小原型出发围绕具体的业务需求在安全性、可靠性、可观测性和成本控制四个维度上不断深化才能真正让 AI 能力可靠地普及到各类工作流中。下一步你可以尝试集成一个真实的 API如天气查询或者为 Agent 添加从向量数据库检索知识的能力。