1. 项目概述当AI学会“组队打怪”如果你最近在捣鼓AI应用尤其是想让大语言模型LLM不只是回答你的问题而是能像一个真正的“智能体”一样自主规划、调用工具、完成任务那么你大概率已经听说过“智能体”这个概念。这不再是科幻电影里的情节而是我们手边就能开始实践的工程。今天要聊的就是OpenAI官方推出的一个重量级工具包openai-agents-python。简单来说这个库是OpenAI为开发者提供的一套“乐高积木”专门用来构建、管理和运行基于其模型的AI智能体。它不是一个成品应用而是一个功能强大的智能体开发框架。想象一下你有一个需求让AI帮你分析一份财报然后根据分析结果自动生成一份投资建议邮件。传统的方式可能是你手动调用多个API自己写逻辑串联。而用这个框架你可以创建一个“分析师”智能体它内部可能包含一个“数据解析”子智能体和“邮件撰写”子智能体它们能自动协作完成整个工作流。这个框架的核心价值就是标准化和简化了多智能体协作的复杂性。它适合谁呢首先是那些已经熟悉OpenAI API并希望将单次对话交互升级为复杂、多步骤自动化流程的开发者。其次是研究多智能体系统MAS的研究者或学生这个库提供了一个生产级的参考实现。最后对于任何想探索AI自动化边界的技术爱好者这都是一个绝佳的起点。2. 核心架构与设计哲学拆解2.1 从“对话”到“智能体”思维的转变在使用传统的openaiPython库时我们的思维模式是“请求-响应”。我们构造一个消息列表messages发送给API然后等待一个回复。整个过程是同步的、被动的。openai-agents-python引入的是一种“智能体”思维。在这里智能体是一个拥有状态、可以执行动作、并能根据环境反馈进行学习的自治实体。框架将一次复杂的任务分解为多个“回合”turns。在每个回合中智能体观察接收当前的输入用户指令、上一步的结果、环境状态。思考基于其内部逻辑通常是LLM决定下一步做什么。行动执行决定比如调用一个工具函数、与另一个智能体通信或者直接生成回复。学习可选根据行动结果更新内部状态或知识。这种模式使得智能体能够处理远超单次对话上下文长度的任务并能整合外部工具和能力。2.2 框架的核心组件Runner, Agent, 与 Tool框架的设计非常清晰主要围绕三个核心抽象Runner运行器这是整个系统的“发动机”和“调度中心”。它负责管理智能体执行的生命周期处理智能体之间的消息传递协调并行或串行执行并最终收集结果。你可以把它想象成一个多线程的“舞台导演”告诉每个“演员”智能体何时上场、说什么台词、听谁的反馈。Runner是实际执行run()方法的核心对象。Agent智能体这是任务的具体执行者。每个智能体都有明确的角色定义通过系统提示词、可用的工具列表以及背后的LLM模型。框架支持多种智能体类型例如OpenAIAgent最基础的智能体直接封装了OpenAI的聊天补全接口可以配置工具。ReActAgent实现了经典的“思考-行动-观察”Reasoning and Acting循环的智能体更适合需要多步推理和工具调用的复杂任务。ToolCallingAgent专门优化了工具调用能力的智能体。Tool工具这是智能体能力的延伸。一个工具本质上是一个Python函数附带清晰的名称和描述。智能体在思考过程中如果认为需要调用工具就会生成相应的参数并“使用”它。框架使得将任意Python函数“包装”成智能体可用的工具变得极其简单。例如你可以创建一个search_web工具、一个execute_sql工具或一个send_email工具。设计哲学这个框架遵循了“约定优于配置”和“显式优于隐式”的原则。它通过清晰的接口和预定义的执行模式如ReAct降低了构建可靠智能体系统的门槛同时保留了足够的灵活性让开发者定制核心逻辑。注意虽然框架提供了便利但构建一个真正鲁棒的智能体系统其挑战从“如何调用API”转移到了“如何设计有效的提示词”、“如何定义清晰的工具接口”以及“如何管理智能体间的协作与冲突”。这是心智模型上的一次重要升级。3. 环境搭建与核心配置详解3.1 基础环境准备开始之前确保你有一个可用的Python环境建议3.9以上和OpenAI API密钥。# 1. 创建并进入项目目录 mkdir my-ai-agent-project cd my-ai-agent-project # 2. 创建虚拟环境强烈推荐 python -m venv venv # 在Windows上激活 # venv\Scripts\activate # 在Mac/Linux上激活 # source venv/bin/activate # 3. 安装核心包 pip install openai-agents安装命令是pip install openai-agents而不是openai-agents-python后者是GitHub仓库名。这个包会依赖openai等必要库。接下来设置你的API密钥。永远不要将密钥硬编码在代码中。最佳实践是使用环境变量。# 在终端中设置临时 export OPENAI_API_KEYsk-your-secret-key-here # Windows (Cmd): set OPENAI_API_KEYsk-your-secret-key-here # Windows (PowerShell): $env:OPENAI_API_KEYsk-your-secret-key-here在你的Python代码中通常不需要显式设置因为openai库会自动从环境变量读取。但如果你想在代码中指定例如使用多个密钥可以这样做from openai import OpenAI client OpenAI(api_keyyour-key) # 不推荐在代码中写死3.2 核心对象初始化实战让我们初始化一个最简单的智能体系统所需的核心对象。import asyncio from agents import Agent, Runner from agents.tool import tool from openai import OpenAI # 初始化OpenAI客户端使用环境变量中的密钥 client OpenAI() # 定义一个简单的工具 tool def get_weather(city: str) - str: 获取指定城市的当前天气信息。 # 这里是一个模拟实现。真实场景下你会调用如OpenWeatherMap的API。 weather_data { 北京: 晴15°C微风, 上海: 多云18°C东南风2级, 深圳: 阵雨22°C南风3级, } return weather_data.get(city, f未找到{city}的天气信息。) # 创建一个智能体并赋予它工具 weather_agent Agent( name气象员, instructions你是一个专业的天气查询助手。请根据用户提供的城市名称调用工具查询天气并以友好、清晰的方式回复用户。如果城市不存在请如实告知。, tools[get_weather], modelgpt-4o, # 指定使用的模型 clientclient, ) # 创建运行器 async def main(): result await Runner.run(weather_agent, 今天深圳的天气怎么样) print(result.final_output) if __name__ __main__: asyncio.run(main())关键配置解析model参数这是智能体背后的“大脑”。gpt-4o在推理和工具调用上表现更佳但成本更高gpt-3.5-turbo速度更快、更经济适合简单任务。根据任务复杂度选择。instructions参数这是智能体的“角色设定”和“行为准则”。写得越具体、越清晰智能体的行为就越可控。这是提示词工程的核心。tool装饰器这是框架的魔法所在。它自动将你的Python函数转换为智能体能理解的工具。函数的文档字符串注释至关重要LLM主要靠它来理解工具的用途和参数。实操心得一工具描述的艺术工具函数的文档字符串docstring是智能体能否正确使用它的关键。描述应遵循“动词开头目标明确参数清晰”的原则。差的描述“处理天气。”好的描述“获取指定城市的当前天气状况包括温度、天气现象和风力。城市名称必须是明确的中文或英文城市名。”好的描述能极大减少智能体的误解和错误调用。4. 构建你的第一个多智能体系统单智能体只是开始多智能体协作才是这个框架威力所在。让我们构建一个简单的“旅行规划”系统包含两个智能体一个信息搜集员和一个行程规划师。4.1 定义角色与工具首先我们为两个智能体定义专属工具。from agents import Agent, Runner from agents.tool import tool from openai import OpenAI import asyncio client OpenAI() # 工具1信息搜集员的工具 - 模拟搜索景点 tool def search_attractions(city: str, interest: str 文化) - str: 根据城市和兴趣偏好如‘文化’、‘美食’、‘自然’搜索推荐景点。 # 模拟数据 data { (北京, 文化): 故宫、天坛、颐和园、国家博物馆, (北京, 美食): 全聚德烤鸭、东来顺涮肉、护国寺小吃街, (上海, 文化): 外滩、豫园、上海博物馆、中共一大会址, (上海, 自然): 上海植物园、世纪公园、崇明东滩湿地, } return data.get((city, interest), f未找到{city}关于{interest}的景点信息。) # 工具2行程规划师的工具 - 生成日程安排 tool def generate_itinerary(attractions: str, days: int) - str: 根据提供的景点列表和旅行天数生成一份详细的每日行程安排。 # 这是一个简单的逻辑实际应用中可能会更复杂甚至调用另一个LLM attraction_list attractions.split(、) itinerary f{days}天行程建议\n for i in range(days): day_num i 1 # 简单分配景点实际可根据距离、开放时间优化 assigned attraction_list[i*2:(i1)*2] if i*2 len(attraction_list) else [f“自由活动或深度体验”] itinerary f第{day_num}天: 参观 {‘、’.join(assigned)}\n return itinerary4.2 创建智能体并设计协作流程现在我们创建两个智能体并设计它们如何协作。我们将使用Runner的run方法串联它们。# 创建信息搜集员智能体 researcher Agent( name旅行信息研究员, instructions你负责根据用户提出的目的地和兴趣搜索相关的旅游景点。你只负责搜索和提供景点名称列表不要进行行程规划。回复应简洁、准确。, tools[search_attractions], modelgpt-3.5-turbo, clientclient, ) # 创建行程规划师智能体 planner Agent( name资深行程规划师, instructions你负责根据研究员提供的景点列表和用户指定的旅行天数制定一份详细、合理、体验丰富的每日行程计划。你的回复需要结构化清晰易读。, tools[generate_itinerary], modelgpt-4o, # 规划需要更强推理使用GPT-4 clientclient, ) # 设计协作流程先研究再规划 async def collaborative_travel_plan(city: str, interest: str, days: int): print(f“用户需求去{city}进行{interest}主题游共{days}天。”) # 第一步研究员搜集信息 print(“\n--- 研究员开始工作 ---”) research_result await Runner.run( researcher, f“请为我搜索{city}关于{interest}的旅游景点。” ) attractions research_result.final_output print(f“研究员找到的景点{attractions}”) # 第二步规划师制定行程 print(“\n--- 规划师开始工作 ---”) # 注意这里我们把上一步的结果attractions和新的指令一起传给规划师 plan_result await Runner.run( planner, f“根据以下景点{attractions}为我规划一个{days}天的详细行程。” ) final_plan plan_result.final_output print(f“\n--- 最终行程计划 ---\n{final_plan}”) return final_plan # 运行协作系统 async def main(): plan await collaborative_travel_plan(“上海”, “文化”, 3) if __name__ “__main__”: asyncio.run(main())这个例子展示了显式多智能体工作流。我们手动控制了智能体的执行顺序和数据传递。Runner.run每次调用都是一个独立的智能体“回合”。这种模式简单直观适用于有明确先后依赖关系的任务。实操心得二智能体间的数据传递在多智能体系统中数据传递是关键。上例中我们将research_result.final_output作为字符串传递。对于更复杂的数据如列表、字典你需要确保它们能被安全地序列化和解析。一种好做法是让智能体以结构化格式如JSON输出下一个智能体再解析。框架本身也支持更复杂的消息传递机制这涉及到AgentContext和Message对象的使用。5. 深入高级特性ReAct模式与并行执行5.1 实现真正的“思考-行动”循环ReActAgent前面的OpenAIAgent在调用工具时相对直接。而ReActAgent实现了更经典的ReAct范式其内部对话会显式包含“Thought:”、“Action:”、“Observation:”等步骤使得推理过程对开发者更透明也常能处理更复杂的多步工具调用。from agents import ReActAgent # 创建一个使用ReAct模式的智能体 react_agent ReActAgent( name复杂问题解决者, instructions你是一个谨慎的解决问题专家。对于复杂问题请一步步思考Thought必要时调用工具Action并根据工具结果Observation继续思考直到得出最终答案Answer。, tools[get_weather, search_attractions], # 赋予它多个工具 modelgpt-4o, clientclient, ) async def solve_complex_problem(): result await Runner.run( react_agent, “我计划本周日从北京出发去上海想参观一些文化景点。请先告诉我北京和上海周日的天气再根据天气情况给我一些上海的文化景点建议。” ) # 查看详细的执行步骤在ReAct模式下非常有用 for turn in result.turns: print(f“Turn {turn.turn_number}:”) if turn.messages: for msg in turn.messages: # 消息可能来自用户、智能体或工具 print(f“ [{msg.source}]: {msg.content[:200]}...”) # 打印前200字符 print(f“\n最终答案{result.final_output}”) # asyncio.run(solve_complex_problem())运行这段代码你会在日志中看到智能体内部清晰的思考链“思考北京天气 - 调用天气工具 - 观察结果 - 思考上海天气 - 调用天气工具 - ... - 综合思考给出建议”。这对于调试智能体的决策过程至关重要。5.2 并行执行与团队协作当多个子任务相互独立时并行执行可以大幅提升效率。openai-agents-python框架通过asyncio原生支持异步使得并行化变得简单。假设我们有三个智能体分别负责资料搜集、数据分析和报告润色且搜集和分析可以同时进行。async def parallel_team_work(): # 假设我们已经定义了三个智能体researcher, analyst, editor researcher Agent(name“研究员”, instructions“...”, tools[], model“gpt-3.5-turbo”, clientclient) analyst Agent(name“分析师”, instructions“...”, tools[], model“gpt-4o”, clientclient) editor Agent(name“编辑”, instructions“...”, tools[], model“gpt-4o”, clientclient) query “分析一下新能源汽车市场在2023年的发展趋势。” # 并行执行研究和分析任务 research_task Runner.run(researcher, f“请搜集关于‘{query}’的最新资料和关键数据。”) analysis_task Runner.run(analyst, f“请基于已知信息对‘{query}’进行初步趋势分析。”) # 等待两个并行任务完成 research_result, analysis_result await asyncio.gather(research_task, analysis_task) print(f“研究完成{research_result.final_output[:100]}...”) print(f“分析完成{analysis_result.final_output[:100]}...”) # 串行执行将前两步的结果交给编辑进行整合润色 final_report_result await Runner.run( editor, f“请将以下研究资料和分析结果整合成一份结构完整、语言流畅的简短报告。\n研究资料{research_result.final_output}\n分析结果{analysis_result.final_output}” ) print(f“\n最终报告\n{final_report_result.final_output}”)关键点asyncio.gather()是实现并发的关键。它允许同时发起多个Runner.run协程任务。但需要注意并行调用API会受到OpenAI账户的速率限制RPM/TPM。在生产环境中需要实现适当的限流或队列机制。6. 生产环境部署与优化指南6.1 错误处理与稳定性智能体系统运行在复杂环境中网络错误、API限制、工具异常、智能体“胡言乱语”都可能发生。健壮的错误处理必不可少。import tenacity from openai import APIError, RateLimitError tenacity.retry( stoptenacity.stop_after_attempt(3), # 最大重试3次 waittenacity.wait_exponential(multiplier1, min4, max10), # 指数退避 retrytenacity.retry_if_exception_type((APIError, RateLimitError)), # 只重试特定错误 before_sleeptenacity.before_sleep_log(logger, logging.INFO) ) async def robust_agent_run(agent, input_text): 一个带有重试机制的智能体运行包装函数。 try: result await Runner.run(agent, input_text) return result except RateLimitError as e: logger.warning(f“速率限制触发等待后重试。错误{e}”) raise # 让tenacity捕获并重试 except APIError as e: logger.error(f“OpenAI API错误{e}”) # 可以根据状态码决定是否重试 if e.status_code 500: raise # 服务器错误重试 else: # 客户端错误如无效请求不应重试 return AgentResult(final_output“抱歉处理您的请求时遇到了服务异常。”) except Exception as e: # 捕获其他未知异常如工具函数抛出的错误 logger.exception(f“智能体执行过程中发生未预期错误{e}”) return AgentResult(final_output“系统处理过程中出现意外错误。”) # 在工具函数内部也应进行防御性编程 tool def call_external_api(url: str) - str: 调用一个外部API。 import requests try: response requests.get(url, timeout10) # 设置超时 response.raise_for_status() # 检查HTTP错误 return response.text except requests.exceptions.Timeout: return “请求外部服务超时。” except requests.exceptions.RequestException as e: return f“调用外部API失败{str(e)}”6.2 性能优化与成本控制模型选择策略采用混合模型策略。将轻量级、高并发的任务如信息检索、简单分类交给gpt-3.5-turbo将需要深度推理、规划或创造性的任务如分析、写作、复杂工具调用交给gpt-4o或gpt-4-turbo。可以在不同智能体上配置不同模型。上下文管理智能体的对话历史会消耗令牌tokens。对于长时间运行的智能体需要实现上下文窗口管理策略例如摘要压缩定期将冗长的历史对话总结成一段简短的摘要替换旧历史。选择性记忆只保留与当前任务最相关的几条消息。工具化记忆将需要长期记忆的信息存入外部数据库如向量库智能体在需要时通过工具查询。缓存机制对于相同或相似的查询其结果往往是相同的。可以为智能体的输出特别是那些调用昂贵外部API或复杂计算的部分实现缓存。可以使用functools.lru_cache装饰简单的工具函数或使用Redis等外部缓存存储复杂响应。监控与日志记录每个智能体调用的输入、输出、使用的令牌数、耗时和调用的工具。这不仅是排查问题的依据也是进行成本分析和性能优化的基础数据。可以考虑集成像LangSmith虽然来自LangChain生态但理念相通这样的专门用于LLM应用监控的平台。6.3 安全与可控性考量工具权限管控不是所有智能体都需要所有工具。一个处理用户反馈的智能体不应该有“删除数据库”工具的访问权限。在定义智能体时严格遵循最小权限原则只授予其完成任务所必需的工具。输入输出过滤与审查对用户输入进行基本的清理和过滤防止提示词注入攻击。对智能体的输出特别是当它被用于执行后续操作如发送邮件、发布内容时应加入人工审核环节或至少是高风险操作确认机制。设置“紧急制动”对于长时间运行或可能进入循环的智能体在Runner层面设置超时timeout和最大回合数max_turns限制。from agents import RunConfig async def safe_run(): config RunConfig( timeout30.0, # 整个运行过程超时30秒 max_turns10, # 最多进行10个回合思考-行动循环 ) result await Runner.run(agent, “复杂任务...”, configconfig)7. 常见问题与排查技巧实录在实际开发和调试openai-agents-python应用时你会遇到一些典型问题。以下是我踩过坑后总结的排查清单。问题现象可能原因排查步骤与解决方案智能体不调用工具直接回答1. 工具描述不清LLM不理解何时用。2. 系统指令instructions未强调使用工具。3. 模型能力不足如用gpt-3.5-turbo处理复杂工具调用。1.检查工具描述确保docstring清晰、具体包含关键词。用“获取XX”、“计算XX”、“查询XX”开头。2.强化指令在instructions中加入“你必须使用我提供的工具来获取信息”或“在回答前先考虑是否需要调用工具”。3.升级模型尝试切换到gpt-4o或gpt-4-turbo它们在工具调用上更可靠。工具调用参数错误1. 工具函数参数类型注解与LLM推断不匹配。2. LLM从用户问题中提取的参数值有歧义。1.使用Pydantic模型对于复杂参数使用tool装饰器支持PydanticBaseModel作为参数这能提供极强的模式校验和说明。2.提供示例在instructions中给智能体举一个正确调用该工具的例子。3.后置校验在工具函数内部对参数进行二次验证和清洗。智能体陷入循环或跑题1. 任务目标不明确。2. 上下文历史过长或包含误导信息。3. 最大回合数设置过高。1.明确终止条件在指令中写明“当你得到最终答案X时你的任务就完成了直接输出X”。2.管理上下文启用RunConfig的message_history_trimming策略或手动在关键步骤后清空/总结历史。3.设置限制使用RunConfig(max_turns5)强制限制循环次数。并行调用时触发速率限制同时发起的API请求超过OpenAI账户的每分钟请求数RPM或令牌数TPM限制。1.实现限流队列使用asyncio.Semaphore或第三方库如aiolimiter控制并发数。2.指数退避重试使用tenacity库为API调用添加重试逻辑捕获RateLimitError并等待。3.监控用量定期检查OpenAI控制台的用量仪表盘根据限制调整并发策略。异步运行时出现奇怪错误在非异步环境如普通脚本中直接调用await Runner.run或在Jupyter Notebook中未正确处理事件循环。1.确保入口点是异步的主函数用async def定义并用asyncio.run(main())调用。2.在Jupyter中使用await直接运行或使用nest_asyncio应用补丁不推荐长期方案。3.检查依赖确保所有自定义工具函数都是线程安全/协程安全的。独家避坑技巧调试利器——打开详细日志框架内置了详细的日志记录。在开发阶段打开它可以看到智能体内部思考、工具调用和消息传递的完整过程对调试有奇效。import logging # 设置agents相关模块的日志级别为DEBUG logging.basicConfig(levellogging.DEBUG) # 或者更精确地控制 logger logging.getLogger(“agents”) logger.setLevel(logging.DEBUG)运行你的代码你会在控制台看到类似[THOUGHT],[ACTION],[OBSERVATION]的日志行让你仿佛拥有了“读心术”能看清智能体的每一步决策。这在排查“智能体为什么这么做”时比猜测要高效一百倍。构建基于openai-agents-python的智能体系统是一个将模糊的AI想法转化为具体、可运行代码的过程。它要求开发者不仅是一个程序员还要成为一个“智能体架构师”思考如何分解任务、设计角色、定义工具和管理协作。这个框架提供的是一套强大的基础设施而真正的魔法来自于你如何用它来构建解决实际问题的智能体团队。从今天开始尝试让你的AI助手从“单兵作战”升级为“集团军协同”你会发现自动化所能触及的边界远比想象中更远。
OpenAI智能体框架实战：从单智能体到多智能体协作开发
1. 项目概述当AI学会“组队打怪”如果你最近在捣鼓AI应用尤其是想让大语言模型LLM不只是回答你的问题而是能像一个真正的“智能体”一样自主规划、调用工具、完成任务那么你大概率已经听说过“智能体”这个概念。这不再是科幻电影里的情节而是我们手边就能开始实践的工程。今天要聊的就是OpenAI官方推出的一个重量级工具包openai-agents-python。简单来说这个库是OpenAI为开发者提供的一套“乐高积木”专门用来构建、管理和运行基于其模型的AI智能体。它不是一个成品应用而是一个功能强大的智能体开发框架。想象一下你有一个需求让AI帮你分析一份财报然后根据分析结果自动生成一份投资建议邮件。传统的方式可能是你手动调用多个API自己写逻辑串联。而用这个框架你可以创建一个“分析师”智能体它内部可能包含一个“数据解析”子智能体和“邮件撰写”子智能体它们能自动协作完成整个工作流。这个框架的核心价值就是标准化和简化了多智能体协作的复杂性。它适合谁呢首先是那些已经熟悉OpenAI API并希望将单次对话交互升级为复杂、多步骤自动化流程的开发者。其次是研究多智能体系统MAS的研究者或学生这个库提供了一个生产级的参考实现。最后对于任何想探索AI自动化边界的技术爱好者这都是一个绝佳的起点。2. 核心架构与设计哲学拆解2.1 从“对话”到“智能体”思维的转变在使用传统的openaiPython库时我们的思维模式是“请求-响应”。我们构造一个消息列表messages发送给API然后等待一个回复。整个过程是同步的、被动的。openai-agents-python引入的是一种“智能体”思维。在这里智能体是一个拥有状态、可以执行动作、并能根据环境反馈进行学习的自治实体。框架将一次复杂的任务分解为多个“回合”turns。在每个回合中智能体观察接收当前的输入用户指令、上一步的结果、环境状态。思考基于其内部逻辑通常是LLM决定下一步做什么。行动执行决定比如调用一个工具函数、与另一个智能体通信或者直接生成回复。学习可选根据行动结果更新内部状态或知识。这种模式使得智能体能够处理远超单次对话上下文长度的任务并能整合外部工具和能力。2.2 框架的核心组件Runner, Agent, 与 Tool框架的设计非常清晰主要围绕三个核心抽象Runner运行器这是整个系统的“发动机”和“调度中心”。它负责管理智能体执行的生命周期处理智能体之间的消息传递协调并行或串行执行并最终收集结果。你可以把它想象成一个多线程的“舞台导演”告诉每个“演员”智能体何时上场、说什么台词、听谁的反馈。Runner是实际执行run()方法的核心对象。Agent智能体这是任务的具体执行者。每个智能体都有明确的角色定义通过系统提示词、可用的工具列表以及背后的LLM模型。框架支持多种智能体类型例如OpenAIAgent最基础的智能体直接封装了OpenAI的聊天补全接口可以配置工具。ReActAgent实现了经典的“思考-行动-观察”Reasoning and Acting循环的智能体更适合需要多步推理和工具调用的复杂任务。ToolCallingAgent专门优化了工具调用能力的智能体。Tool工具这是智能体能力的延伸。一个工具本质上是一个Python函数附带清晰的名称和描述。智能体在思考过程中如果认为需要调用工具就会生成相应的参数并“使用”它。框架使得将任意Python函数“包装”成智能体可用的工具变得极其简单。例如你可以创建一个search_web工具、一个execute_sql工具或一个send_email工具。设计哲学这个框架遵循了“约定优于配置”和“显式优于隐式”的原则。它通过清晰的接口和预定义的执行模式如ReAct降低了构建可靠智能体系统的门槛同时保留了足够的灵活性让开发者定制核心逻辑。注意虽然框架提供了便利但构建一个真正鲁棒的智能体系统其挑战从“如何调用API”转移到了“如何设计有效的提示词”、“如何定义清晰的工具接口”以及“如何管理智能体间的协作与冲突”。这是心智模型上的一次重要升级。3. 环境搭建与核心配置详解3.1 基础环境准备开始之前确保你有一个可用的Python环境建议3.9以上和OpenAI API密钥。# 1. 创建并进入项目目录 mkdir my-ai-agent-project cd my-ai-agent-project # 2. 创建虚拟环境强烈推荐 python -m venv venv # 在Windows上激活 # venv\Scripts\activate # 在Mac/Linux上激活 # source venv/bin/activate # 3. 安装核心包 pip install openai-agents安装命令是pip install openai-agents而不是openai-agents-python后者是GitHub仓库名。这个包会依赖openai等必要库。接下来设置你的API密钥。永远不要将密钥硬编码在代码中。最佳实践是使用环境变量。# 在终端中设置临时 export OPENAI_API_KEYsk-your-secret-key-here # Windows (Cmd): set OPENAI_API_KEYsk-your-secret-key-here # Windows (PowerShell): $env:OPENAI_API_KEYsk-your-secret-key-here在你的Python代码中通常不需要显式设置因为openai库会自动从环境变量读取。但如果你想在代码中指定例如使用多个密钥可以这样做from openai import OpenAI client OpenAI(api_keyyour-key) # 不推荐在代码中写死3.2 核心对象初始化实战让我们初始化一个最简单的智能体系统所需的核心对象。import asyncio from agents import Agent, Runner from agents.tool import tool from openai import OpenAI # 初始化OpenAI客户端使用环境变量中的密钥 client OpenAI() # 定义一个简单的工具 tool def get_weather(city: str) - str: 获取指定城市的当前天气信息。 # 这里是一个模拟实现。真实场景下你会调用如OpenWeatherMap的API。 weather_data { 北京: 晴15°C微风, 上海: 多云18°C东南风2级, 深圳: 阵雨22°C南风3级, } return weather_data.get(city, f未找到{city}的天气信息。) # 创建一个智能体并赋予它工具 weather_agent Agent( name气象员, instructions你是一个专业的天气查询助手。请根据用户提供的城市名称调用工具查询天气并以友好、清晰的方式回复用户。如果城市不存在请如实告知。, tools[get_weather], modelgpt-4o, # 指定使用的模型 clientclient, ) # 创建运行器 async def main(): result await Runner.run(weather_agent, 今天深圳的天气怎么样) print(result.final_output) if __name__ __main__: asyncio.run(main())关键配置解析model参数这是智能体背后的“大脑”。gpt-4o在推理和工具调用上表现更佳但成本更高gpt-3.5-turbo速度更快、更经济适合简单任务。根据任务复杂度选择。instructions参数这是智能体的“角色设定”和“行为准则”。写得越具体、越清晰智能体的行为就越可控。这是提示词工程的核心。tool装饰器这是框架的魔法所在。它自动将你的Python函数转换为智能体能理解的工具。函数的文档字符串注释至关重要LLM主要靠它来理解工具的用途和参数。实操心得一工具描述的艺术工具函数的文档字符串docstring是智能体能否正确使用它的关键。描述应遵循“动词开头目标明确参数清晰”的原则。差的描述“处理天气。”好的描述“获取指定城市的当前天气状况包括温度、天气现象和风力。城市名称必须是明确的中文或英文城市名。”好的描述能极大减少智能体的误解和错误调用。4. 构建你的第一个多智能体系统单智能体只是开始多智能体协作才是这个框架威力所在。让我们构建一个简单的“旅行规划”系统包含两个智能体一个信息搜集员和一个行程规划师。4.1 定义角色与工具首先我们为两个智能体定义专属工具。from agents import Agent, Runner from agents.tool import tool from openai import OpenAI import asyncio client OpenAI() # 工具1信息搜集员的工具 - 模拟搜索景点 tool def search_attractions(city: str, interest: str 文化) - str: 根据城市和兴趣偏好如‘文化’、‘美食’、‘自然’搜索推荐景点。 # 模拟数据 data { (北京, 文化): 故宫、天坛、颐和园、国家博物馆, (北京, 美食): 全聚德烤鸭、东来顺涮肉、护国寺小吃街, (上海, 文化): 外滩、豫园、上海博物馆、中共一大会址, (上海, 自然): 上海植物园、世纪公园、崇明东滩湿地, } return data.get((city, interest), f未找到{city}关于{interest}的景点信息。) # 工具2行程规划师的工具 - 生成日程安排 tool def generate_itinerary(attractions: str, days: int) - str: 根据提供的景点列表和旅行天数生成一份详细的每日行程安排。 # 这是一个简单的逻辑实际应用中可能会更复杂甚至调用另一个LLM attraction_list attractions.split(、) itinerary f{days}天行程建议\n for i in range(days): day_num i 1 # 简单分配景点实际可根据距离、开放时间优化 assigned attraction_list[i*2:(i1)*2] if i*2 len(attraction_list) else [f“自由活动或深度体验”] itinerary f第{day_num}天: 参观 {‘、’.join(assigned)}\n return itinerary4.2 创建智能体并设计协作流程现在我们创建两个智能体并设计它们如何协作。我们将使用Runner的run方法串联它们。# 创建信息搜集员智能体 researcher Agent( name旅行信息研究员, instructions你负责根据用户提出的目的地和兴趣搜索相关的旅游景点。你只负责搜索和提供景点名称列表不要进行行程规划。回复应简洁、准确。, tools[search_attractions], modelgpt-3.5-turbo, clientclient, ) # 创建行程规划师智能体 planner Agent( name资深行程规划师, instructions你负责根据研究员提供的景点列表和用户指定的旅行天数制定一份详细、合理、体验丰富的每日行程计划。你的回复需要结构化清晰易读。, tools[generate_itinerary], modelgpt-4o, # 规划需要更强推理使用GPT-4 clientclient, ) # 设计协作流程先研究再规划 async def collaborative_travel_plan(city: str, interest: str, days: int): print(f“用户需求去{city}进行{interest}主题游共{days}天。”) # 第一步研究员搜集信息 print(“\n--- 研究员开始工作 ---”) research_result await Runner.run( researcher, f“请为我搜索{city}关于{interest}的旅游景点。” ) attractions research_result.final_output print(f“研究员找到的景点{attractions}”) # 第二步规划师制定行程 print(“\n--- 规划师开始工作 ---”) # 注意这里我们把上一步的结果attractions和新的指令一起传给规划师 plan_result await Runner.run( planner, f“根据以下景点{attractions}为我规划一个{days}天的详细行程。” ) final_plan plan_result.final_output print(f“\n--- 最终行程计划 ---\n{final_plan}”) return final_plan # 运行协作系统 async def main(): plan await collaborative_travel_plan(“上海”, “文化”, 3) if __name__ “__main__”: asyncio.run(main())这个例子展示了显式多智能体工作流。我们手动控制了智能体的执行顺序和数据传递。Runner.run每次调用都是一个独立的智能体“回合”。这种模式简单直观适用于有明确先后依赖关系的任务。实操心得二智能体间的数据传递在多智能体系统中数据传递是关键。上例中我们将research_result.final_output作为字符串传递。对于更复杂的数据如列表、字典你需要确保它们能被安全地序列化和解析。一种好做法是让智能体以结构化格式如JSON输出下一个智能体再解析。框架本身也支持更复杂的消息传递机制这涉及到AgentContext和Message对象的使用。5. 深入高级特性ReAct模式与并行执行5.1 实现真正的“思考-行动”循环ReActAgent前面的OpenAIAgent在调用工具时相对直接。而ReActAgent实现了更经典的ReAct范式其内部对话会显式包含“Thought:”、“Action:”、“Observation:”等步骤使得推理过程对开发者更透明也常能处理更复杂的多步工具调用。from agents import ReActAgent # 创建一个使用ReAct模式的智能体 react_agent ReActAgent( name复杂问题解决者, instructions你是一个谨慎的解决问题专家。对于复杂问题请一步步思考Thought必要时调用工具Action并根据工具结果Observation继续思考直到得出最终答案Answer。, tools[get_weather, search_attractions], # 赋予它多个工具 modelgpt-4o, clientclient, ) async def solve_complex_problem(): result await Runner.run( react_agent, “我计划本周日从北京出发去上海想参观一些文化景点。请先告诉我北京和上海周日的天气再根据天气情况给我一些上海的文化景点建议。” ) # 查看详细的执行步骤在ReAct模式下非常有用 for turn in result.turns: print(f“Turn {turn.turn_number}:”) if turn.messages: for msg in turn.messages: # 消息可能来自用户、智能体或工具 print(f“ [{msg.source}]: {msg.content[:200]}...”) # 打印前200字符 print(f“\n最终答案{result.final_output}”) # asyncio.run(solve_complex_problem())运行这段代码你会在日志中看到智能体内部清晰的思考链“思考北京天气 - 调用天气工具 - 观察结果 - 思考上海天气 - 调用天气工具 - ... - 综合思考给出建议”。这对于调试智能体的决策过程至关重要。5.2 并行执行与团队协作当多个子任务相互独立时并行执行可以大幅提升效率。openai-agents-python框架通过asyncio原生支持异步使得并行化变得简单。假设我们有三个智能体分别负责资料搜集、数据分析和报告润色且搜集和分析可以同时进行。async def parallel_team_work(): # 假设我们已经定义了三个智能体researcher, analyst, editor researcher Agent(name“研究员”, instructions“...”, tools[], model“gpt-3.5-turbo”, clientclient) analyst Agent(name“分析师”, instructions“...”, tools[], model“gpt-4o”, clientclient) editor Agent(name“编辑”, instructions“...”, tools[], model“gpt-4o”, clientclient) query “分析一下新能源汽车市场在2023年的发展趋势。” # 并行执行研究和分析任务 research_task Runner.run(researcher, f“请搜集关于‘{query}’的最新资料和关键数据。”) analysis_task Runner.run(analyst, f“请基于已知信息对‘{query}’进行初步趋势分析。”) # 等待两个并行任务完成 research_result, analysis_result await asyncio.gather(research_task, analysis_task) print(f“研究完成{research_result.final_output[:100]}...”) print(f“分析完成{analysis_result.final_output[:100]}...”) # 串行执行将前两步的结果交给编辑进行整合润色 final_report_result await Runner.run( editor, f“请将以下研究资料和分析结果整合成一份结构完整、语言流畅的简短报告。\n研究资料{research_result.final_output}\n分析结果{analysis_result.final_output}” ) print(f“\n最终报告\n{final_report_result.final_output}”)关键点asyncio.gather()是实现并发的关键。它允许同时发起多个Runner.run协程任务。但需要注意并行调用API会受到OpenAI账户的速率限制RPM/TPM。在生产环境中需要实现适当的限流或队列机制。6. 生产环境部署与优化指南6.1 错误处理与稳定性智能体系统运行在复杂环境中网络错误、API限制、工具异常、智能体“胡言乱语”都可能发生。健壮的错误处理必不可少。import tenacity from openai import APIError, RateLimitError tenacity.retry( stoptenacity.stop_after_attempt(3), # 最大重试3次 waittenacity.wait_exponential(multiplier1, min4, max10), # 指数退避 retrytenacity.retry_if_exception_type((APIError, RateLimitError)), # 只重试特定错误 before_sleeptenacity.before_sleep_log(logger, logging.INFO) ) async def robust_agent_run(agent, input_text): 一个带有重试机制的智能体运行包装函数。 try: result await Runner.run(agent, input_text) return result except RateLimitError as e: logger.warning(f“速率限制触发等待后重试。错误{e}”) raise # 让tenacity捕获并重试 except APIError as e: logger.error(f“OpenAI API错误{e}”) # 可以根据状态码决定是否重试 if e.status_code 500: raise # 服务器错误重试 else: # 客户端错误如无效请求不应重试 return AgentResult(final_output“抱歉处理您的请求时遇到了服务异常。”) except Exception as e: # 捕获其他未知异常如工具函数抛出的错误 logger.exception(f“智能体执行过程中发生未预期错误{e}”) return AgentResult(final_output“系统处理过程中出现意外错误。”) # 在工具函数内部也应进行防御性编程 tool def call_external_api(url: str) - str: 调用一个外部API。 import requests try: response requests.get(url, timeout10) # 设置超时 response.raise_for_status() # 检查HTTP错误 return response.text except requests.exceptions.Timeout: return “请求外部服务超时。” except requests.exceptions.RequestException as e: return f“调用外部API失败{str(e)}”6.2 性能优化与成本控制模型选择策略采用混合模型策略。将轻量级、高并发的任务如信息检索、简单分类交给gpt-3.5-turbo将需要深度推理、规划或创造性的任务如分析、写作、复杂工具调用交给gpt-4o或gpt-4-turbo。可以在不同智能体上配置不同模型。上下文管理智能体的对话历史会消耗令牌tokens。对于长时间运行的智能体需要实现上下文窗口管理策略例如摘要压缩定期将冗长的历史对话总结成一段简短的摘要替换旧历史。选择性记忆只保留与当前任务最相关的几条消息。工具化记忆将需要长期记忆的信息存入外部数据库如向量库智能体在需要时通过工具查询。缓存机制对于相同或相似的查询其结果往往是相同的。可以为智能体的输出特别是那些调用昂贵外部API或复杂计算的部分实现缓存。可以使用functools.lru_cache装饰简单的工具函数或使用Redis等外部缓存存储复杂响应。监控与日志记录每个智能体调用的输入、输出、使用的令牌数、耗时和调用的工具。这不仅是排查问题的依据也是进行成本分析和性能优化的基础数据。可以考虑集成像LangSmith虽然来自LangChain生态但理念相通这样的专门用于LLM应用监控的平台。6.3 安全与可控性考量工具权限管控不是所有智能体都需要所有工具。一个处理用户反馈的智能体不应该有“删除数据库”工具的访问权限。在定义智能体时严格遵循最小权限原则只授予其完成任务所必需的工具。输入输出过滤与审查对用户输入进行基本的清理和过滤防止提示词注入攻击。对智能体的输出特别是当它被用于执行后续操作如发送邮件、发布内容时应加入人工审核环节或至少是高风险操作确认机制。设置“紧急制动”对于长时间运行或可能进入循环的智能体在Runner层面设置超时timeout和最大回合数max_turns限制。from agents import RunConfig async def safe_run(): config RunConfig( timeout30.0, # 整个运行过程超时30秒 max_turns10, # 最多进行10个回合思考-行动循环 ) result await Runner.run(agent, “复杂任务...”, configconfig)7. 常见问题与排查技巧实录在实际开发和调试openai-agents-python应用时你会遇到一些典型问题。以下是我踩过坑后总结的排查清单。问题现象可能原因排查步骤与解决方案智能体不调用工具直接回答1. 工具描述不清LLM不理解何时用。2. 系统指令instructions未强调使用工具。3. 模型能力不足如用gpt-3.5-turbo处理复杂工具调用。1.检查工具描述确保docstring清晰、具体包含关键词。用“获取XX”、“计算XX”、“查询XX”开头。2.强化指令在instructions中加入“你必须使用我提供的工具来获取信息”或“在回答前先考虑是否需要调用工具”。3.升级模型尝试切换到gpt-4o或gpt-4-turbo它们在工具调用上更可靠。工具调用参数错误1. 工具函数参数类型注解与LLM推断不匹配。2. LLM从用户问题中提取的参数值有歧义。1.使用Pydantic模型对于复杂参数使用tool装饰器支持PydanticBaseModel作为参数这能提供极强的模式校验和说明。2.提供示例在instructions中给智能体举一个正确调用该工具的例子。3.后置校验在工具函数内部对参数进行二次验证和清洗。智能体陷入循环或跑题1. 任务目标不明确。2. 上下文历史过长或包含误导信息。3. 最大回合数设置过高。1.明确终止条件在指令中写明“当你得到最终答案X时你的任务就完成了直接输出X”。2.管理上下文启用RunConfig的message_history_trimming策略或手动在关键步骤后清空/总结历史。3.设置限制使用RunConfig(max_turns5)强制限制循环次数。并行调用时触发速率限制同时发起的API请求超过OpenAI账户的每分钟请求数RPM或令牌数TPM限制。1.实现限流队列使用asyncio.Semaphore或第三方库如aiolimiter控制并发数。2.指数退避重试使用tenacity库为API调用添加重试逻辑捕获RateLimitError并等待。3.监控用量定期检查OpenAI控制台的用量仪表盘根据限制调整并发策略。异步运行时出现奇怪错误在非异步环境如普通脚本中直接调用await Runner.run或在Jupyter Notebook中未正确处理事件循环。1.确保入口点是异步的主函数用async def定义并用asyncio.run(main())调用。2.在Jupyter中使用await直接运行或使用nest_asyncio应用补丁不推荐长期方案。3.检查依赖确保所有自定义工具函数都是线程安全/协程安全的。独家避坑技巧调试利器——打开详细日志框架内置了详细的日志记录。在开发阶段打开它可以看到智能体内部思考、工具调用和消息传递的完整过程对调试有奇效。import logging # 设置agents相关模块的日志级别为DEBUG logging.basicConfig(levellogging.DEBUG) # 或者更精确地控制 logger logging.getLogger(“agents”) logger.setLevel(logging.DEBUG)运行你的代码你会在控制台看到类似[THOUGHT],[ACTION],[OBSERVATION]的日志行让你仿佛拥有了“读心术”能看清智能体的每一步决策。这在排查“智能体为什么这么做”时比猜测要高效一百倍。构建基于openai-agents-python的智能体系统是一个将模糊的AI想法转化为具体、可运行代码的过程。它要求开发者不仅是一个程序员还要成为一个“智能体架构师”思考如何分解任务、设计角色、定义工具和管理协作。这个框架提供的是一套强大的基础设施而真正的魔法来自于你如何用它来构建解决实际问题的智能体团队。从今天开始尝试让你的AI助手从“单兵作战”升级为“集团军协同”你会发现自动化所能触及的边界远比想象中更远。
