大家好,我是男秀。这一期和大家聊聊 AI 智能体(AI Agent)。
在实际开发 AI 应用的过程中,我们常常会碰到一些复杂任务,它们无法仅靠一次和大模型的简单对话完成。比如生成一份包含多步骤的市场报告、自动处理数据并保存结果,或是协调多个 AI 角色共同完成一项任务。这类需求往往需要一个连贯的工作流程,并且可能需要调用外部工具——而这正是 AI 智能体所擅长的。
虽然 LangChain 在这方面很强大,但对于想快速实现上述需求的开发者来说,它有太多的概念需要学习了。我一直希望找到一个更简单直接的框架,直到我深入接触了 OpenAI 官方推出的 openai-agents。它轻量、概念清晰、上手快,对大多数任务来说刚刚好。
智能体的核心概念
那么,openai-agents 是如何实现这些能力的呢?根据官方文档,我总结了三个核心概念。理解了它们,你就掌握了开发 AI 智能体的关键。
第一个是Agent(代理)。它是框架的核心,你可以把它看作一个”AI小助手”。通过配置,我们能定义它的角色、所用的模型和可调用的工具。主要包括:
- 指令(Instructions):描述它的角色和任务,比如”你是一位文案专家”;
- 模型(Model):指定使用哪个大模型,比如gpt-4o-mini;
- 工具(Tools):赋予它调用外部函数或API的能力。比如把结果保存到本地文件中。
第二个概念是 Handoff(交接),这是实现多智能体协作的关键机制。Handoff 本身是一种特殊工具,当一个 Agent 调用它时,就意味着它已完成当前任务,并将”接力棒”及处理结果交给下一个指定的 Agent。这正是构建工作流的基础。
第三个概念是 Tool Calling(工具调用),它使得大模型能够调用开发者提前编写好的函数,来完成特定任务,如查询数据库、读写文件或调用 API。例如,当大模型被要求”把今天的天气保存到文件里”,它不会自己去写文件,而是调用一个名为 save_to_file 的工具来完成操作。这极大扩展了 Agent 的能力边界。
快速上手
如果你使用uv或者pip,可以使用以下命令快速安装openai-agents。
1 2 3 4 5
| uv pip install openai-agents
pip install openai-agents
|
安装完成后,我们可以用下面这段”Hello World”代码快速体验:
1 2 3 4 5
| from agents import Agent, Runner
agent = Agent(name="Assistant", instructions="你是一位有帮助的助手", model="gpt-4o-mini") result = Runner.run_sync(agent, "写一首关于编程中递归的俳句。") print(result.final_output)
|
代码非常简洁,参数也很直观。在创建 Agent 对象时:
name 是智能体的名称,建议使用英文;
instructions 是系统提示词,用于定义其角色;
model 可指定使用的大模型。
如果你需要使用国内支持 OpenAI SDK 格式的大模型,可以在代码开头进行如下配置:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| import os from dotenv import load_dotenv from openai import AsyncOpenAI from agents import set_default_openai_client, set_default_openai_api
load_dotenv()
base_url = os.getenv('OPENAI_BASE_URL') api_key = os.getenv('OPENAI_API_KEY')
custom_client = AsyncOpenAI(base_url=base_url, api_key=api_key) set_default_openai_client(custom_client) set_default_openai_api("chat_completions")
agent = Agent(name="Assistant", instructions="你是一位有帮助的助手", model="gpt-4o-mini")
|
常见代理模式
在掌握了基础用法之后,我们来看几种常见的代理模式如何用 openai-agents 实现。
调用工具模式
这是最基本也是最常用的模式,让 Agent 能够与外部环境互动。例如,我们可以定义一个工具函数,让 Agent 将生成的文案保存为本地文件。
首先,使用 function_tool 装饰器定义一个工具函数:
1 2 3 4 5 6 7 8 9 10
| from agents import function_tool
@function_tool def save_to_file(filename: str, content: str) -> str: try: with open(filename, 'w', encoding='utf-8') as f: f.write(content) return f"成功将内容保存到文件: {filename}" except Exception as e: return f"保存文件失败: {e}"
|
然后定义一个文件处理助手 Agent,并将工具传入:
1 2 3 4 5 6
| file_saver_agent = Agent( name="文件处理助手", instructions="你是一个能写诗,并且能将诗歌保存到文件的助手。", model='gpt-4o-mini', tools=[save_to_file], )
|
最后调用该 Agent:
1 2 3 4 5
| prompt = "写一首关于夏日夜晚的短诗,并将其保存到名为 'summer_night.txt' 的文件中。" result = Runner.run_sync(file_saver_agent, prompt)
print(result.final_output)
|
执行完成后,你可以在项目目录中找到新生成的 summer_night.txt。
工作流模式
也称为”流水线模式”,适用于多步骤任务。每个步骤由专门的 Agent 处理,上一个的输出作为下一个的输入。
比如我们要写一篇完整的营销文案,可以拆分为三个步骤,对应三个Agent:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
| from agents import Agent
topic_agent = Agent( name="选题策划人", instructions="你是一个爆款选题策划,请为'AI Agent'这个主题想出3个吸引人的标题。", model="gpt-4o-mini", )
outline_agent = Agent( name="大纲规划师", instructions="根据给定的选题,生成一份结构清晰、有逻辑的文章大纲。", model="gpt-4o-mini", )
writer_agent = Agent( name="文案撰写师", instructions="根据提供的大纲,写一篇生动有趣、吸引人的文案。", model="gpt-4o-mini", )
|
按顺序调用它们:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
| from agents import Runner
input_prompt = input("请输入你想要的文案主题:")
topic_result = Runner.run_sync( topic_agent, f"请为'{input_prompt}'这个主题想出3个吸引人的标题。", ) print("=== 选题策划结果 ===") print(topic_result.final_output)
outline_result = Runner.run_sync( outline_agent, f"请根据以下选题生成一份简短的文案大纲,不要超过100字:\n{topic_result.final_output}", ) print("=== 大纲规划结果 ===") print(outline_result.final_output)
writer_result = Runner.run_sync( writer_agent, f"请根据以下大纲写一篇简短的文案,不要超过300字:\n{outline_result.final_output}", ) print("=== 最终文案 ===") print(writer_result.final_output)
|
最终就能实现选题→写大纲→写文案这么一个流程。
路由模式
该模式类似一个”任务调度员”,它本身不处理任务,而是根据内容类型将请求分发给专门的 Agent。
我们用文案生成的例子来说明:假设我们有三个专门的文案生成器,分别用于不同平台。
- 小红书文案生成器
- 口播(抖音/视频号)文案生成器
- 朋友圈文案生成器
比如我们定义三个专业的文案生成Agent:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
| from agents import Agent
xiaohongshu_agent = Agent( name="xiaohongshu_agent", instructions="用小红书风格写一篇种草文案。", model="gpt-4o-mini", )
douyin_agent = Agent( name="douyin_agent", instructions="用口语化的风格写一个引人入胜的视频口播稿。", model="gpt-4o-mini", )
wechat_agent = Agent( name="wechat_agent", instructions="用简洁、有感染力的风格写一段朋友圈文案。", model="gpt-4o-mini", )
|
再定义一个路由 Agent,将上述 Agent 设为可交接对象:
1 2 3 4 5 6 7
| router_agent = Agent( name="文案任务路由器", instructions="根据用户的请求,判断应该将任务交接给哪个文案生成器。不要自己生成文案,只做分发。", model="gpt-4o-mini", handoffs=[xiaohongshu_agent, douyin_agent, wechat_agent], )
|
最后通过循环处理用户输入,实现自动路由:
1 2 3 4 5 6 7 8 9 10 11 12 13
| agent = router_agent msg = input("您好!请告诉我您的需求,我可以帮您生成小红书、抖音或朋友圈文案。") user_messages = [{"content": msg, "role": "user"}]
while True: result = Runner.run_sync(agent, user_messages) print(result.final_output) user_messages = result.to_input_list() print("\n") agent = router_agent user_msg = input(f"当前 Agent 是:{agent.name},请输入消息:") user_messages.append({"content": user_msg, "role": "user"})
|
生成器-评估器模式
这个模式很好理解,就是”一个写,一个改”。通俗地说,是让一个Agent(生成器)生成内容,然后让另一个Agent(评估器)扮演裁判或评论家的角色,去评估内容的质量。
我们定义两个 Agent:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
| from agents import Agent
story_outline_generator = Agent( name="story_outline_generator", instructions=( "你根据用户输入生成一个非常简短的故事大纲。" "如果有反馈,请根据反馈改进大纲。" ), model='gpt-4o-mini', )
evaluator = Agent( name="evaluator", instructions=( "你需要评估一个故事大纲,并判断它是否足够好。" "如果不够好,请给出需要改进的反馈。" "第一次绝不能通过。在5次尝试后,如果大纲足够好可以通过——不必追求完美。" "\n请按以下JSON格式回复:\n" '{"score": "pass|needs_improvement|fail", "feedback": "你的反馈内容"}' ), model='gpt-4o-mini', )
|
设定好Agent后,我们还需要编写解析函数,用于提取评估结果中的评分和反馈:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16
| def parse_evaluation(text: str) -> dict: """解析评估文本,提取评分和反馈""" import json
try: data = json.loads(text.strip()) return { "feedback": data.get("feedback", ""), "score": data.get("score", "needs_improvement") } except json.JSONDecodeError: return { "feedback": "解析失败,请重试", "score": "needs_improvement" }
|
最后是主流程。同样的,我们从获取终端的用户输入,然后组装到user_messages里。然后在一个死循环里面,使用Runner.run方法调用生成故事大纲Agent后,获取其响应值作为评判Agent的输入。循环几次后,当评判Agent通过了,就输出最终的故事大纲。
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36
| msg = input("你想听什么样的故事?") user_messages: list[TResponseInputItem] = [{"content": msg, "role": "user"}]
latest_outline: str | None = None
while True: story_outline_result = Runner.run_sync( story_outline_generator, user_messages, )
user_messages = story_outline_result.to_input_list() latest_outline = ItemHelpers.text_message_outputs( story_outline_result.new_items) print("已生成故事大纲") print(latest_outline)
evaluator_result = Runner.run_sync(evaluator, user_messages) evaluation_text = ItemHelpers.text_message_outputs( evaluator_result.new_items)
result = parse_evaluation(evaluation_text) print(f"裁判评分: {result['score']}") print(f"裁判反馈: {result['feedback']}")
if result['score'] == "pass": print("故事大纲已足够好,流程结束。") break
print("根据反馈重新生成大纲")
user_messages.append( {"content": f"反馈: {result['feedback']}", "role": "user"})
print(f"最终故事大纲: {latest_outline}")
|
总结
这一期我们聊的openai-agents框架,最大的特点就是简单。
它没有引入过多复杂的概念或抽象层,核心组成(Agent、Handoff、Tool)清晰直观,代码写起来也很自然,能让我们快速将想法落地。无论是构建简单的内容生成流水线,还是实现稍复杂的任务分发系统,它都提供了一个轻量而强大的解决方案。
如果你也希望快速实现多智能体协作,又不愿陷入繁杂的配置和概念中,那么 openai-agents 绝对值得一试。