我的碎碎念

不为繁华易匠心,不舍初心得始终。

0%

AI智能体开发从未如此简单:OpenAI官方轻量框架 openai-agents 上手指南

大家好,我是男秀。这一期和大家聊聊 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
uv pip install openai-agents

# 或者使用 pip
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) # 设置openai客户端
set_default_openai_api("chat_completions") # 设置api格式

# 创建Agent
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)

# Agent的最终输出通常是确认信息
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

# 选题 Agent
topic_agent = Agent(
name="选题策划人",
instructions="你是一个爆款选题策划,请为'AI Agent'这个主题想出3个吸引人的标题。",
model="gpt-4o-mini",
)

# 大纲 Agent
outline_agent = Agent(
name="大纲规划师",
instructions="根据给定的选题,生成一份结构清晰、有逻辑的文章大纲。",
model="gpt-4o-mini",
)

# 文案 Agent
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
# 定义路由Agent
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:
# 尝试直接解析JSON
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 绝对值得一试。

欢迎关注我的其它发布渠道