分享开发技术
0%

Openai Agents的介绍和使用

发表于 更新于
本文字数： 4.6k 阅读时长 ≈ 4 分钟

opneai发布了一款agent框架openai-agents，它是一个轻量但功能强大的多智能体框架，这边文章介绍下它的使用。

openai agents的介绍

openai-agents和其它大多数agent框架的功能一样，但是它有以下几个核心概念：

1
2
3
4
代理：LLMs已配置说明、工具、护栏和交接
交接：允许代理将控制权移交给其他代理以完成特定任务
护栏：可配置的输入和输出验证安全检查
跟踪：内置代理运行跟踪功能，允许您查看、调试和优化工作流程

安装：

1
2
pip install openai-agents
pip install 'httpx[socks]'

如果使用openai的模型，需要设置openai api key:

1
export OPENAI_API_KEY=xxx

github地址和相关文档：

1
2
3
https://github.com/openai/openai-agents-python
https://openai.github.io/openai-agents-python/
https://platform.openai.comdocs/guides/agents-sdk

openai agents的使用

创建基础的agent

可以使用 agents.Agent 来创建一个agent，创建好agent后有三种方式运行它:

1
2
3
run_sync(): 以同步非流式的方式运行agent
run(): 以异步非流式的方式运行agent
run_streamed(): 以异步流式的方式运行agent

创建agent并使用同步非流式的方式运行：

1
2
3
4
5
6
7
8
9
10
11
from agents import Agent, Runner


agent = Agent(
    name="Assistant",
    instructions="你是一个能给我提供帮助的人，请回答我下面提出的问题",
    model='gpt-4o-mini'
)

result = Runner.run_sync(agent, "请计算1+1等于几")
print(result.final_output)

创建agent并使用异步非流式的方式运行:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import asyncio
from agents import Agent, Runner


async def main():
    agent = Agent(
        name="Assistant",
        instructions="你是一个能给我提供帮助的人，请回答我下面提出的问题",
        model='gpt-3.5-turbo'
    )

    result = await Runner.run(agent, input="1+1等于几")
    return result


if __name__ == '__main__':
    ret = asyncio.run(main())
    print(ret)

创建agent并使用异步流式的方式运行:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import asyncio
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner


async def main():
    agent = Agent(
        name="Assistant",
        instructions="你是一个能给我提供帮助的人，请回答我下面提出的问题",
    )

    result = Runner.run_streamed(agent, input="1+1等于几")
    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            print(f"event: {event}")
            # print(event.data.delta, end="", flush=True)


if __name__ == '__main__':
    asyncio.run(main())

使用非OpenAI模型创建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
from openai import AsyncOpenAI
from agents import Agent, Runner, OpenAIChatCompletionsModel


BASE_URL = "https://api.deepseek.com"
API_KEY = 'xxx'
MODEL_NAME = 'deepseek-reasoner'
# 定义一个deepseek的大模型客户端
client = AsyncOpenAI(base_url=BASE_URL, api_key=API_KEY)


def main():
    agent = Agent(
        name="Assistant",
        instructions="你是一个能给我提供帮助的人，请回答我下面提出的问题",
        model=OpenAIChatCompletionsModel(model=MODEL_NAME, openai_client=client)
    )

    # 非流式输出
    result = Runner.run_sync(agent, "你是谁,你来自哪里")
    return result.final_output


if __name__ == '__main__':
    ret = main()
    print(ret)

创建工具调用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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import asyncio
from agents import Agent, Runner, function_tool


MODEL = "gpt-4o-mini"


# 定义tool
@function_tool()
def query_weather(city: str, date: str):
    """
    查询城市的天气信息
    :param city: 城市
    :param date: 日期
    :return:
    """
    print(f"查询[{date}] [{city}]的天气信息 ...")

    return {
        "code": 0,
        "data": {
            "city": city,
            "data": date,
            "weather": "sunny",
            "temperature": "18°C",
            "humidity": "80%",
        }
    }


# 定义tool
@function_tool()
def query_stock(stock_code: str, date: str):
    """
    查询股票信息
    :param stock_code: 股票代码
    :param date: 日期
    :return:
    """
    print(f"查询[{date}] [{stock_code}]的涨跌信息 ...")

    return {
        "code": 0,
        "data": {
            "stock_code": stock_code,
            "data": date,
            "max": "10.05",
            "min": "8,02"
        }
    }


smarter = Agent(
    name="smart_man",
    instructions="你是一个无所不知无所不晓的人，上知天文下知地理.",
    model=MODEL,
    tools=[query_weather, query_stock]    # 指定多个tool
)


async def main():
    result = await Runner.run(smarter, input="你好，请帮我查询武汉2025年3月20日的天气信息")
    print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())

创建多个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
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
import asyncio
from agents import Agent, Runner, handoff


MODEL = "gpt-4o-mini"


teacher_math = Agent(
    name="math_teacher",
    instructions="你是一个高中数学老师，你的工作是解答学生的数学问题，制定详细的数学学习计划等",
    model=MODEL
)


teacher_chinese = Agent(
    name="chinese_teacher",
    instructions="你是一个高中语文老师，你的工作是解答学生的语文问题，制定详细的语文学习计划等",
    model=MODEL
)


def teacher_math_handoff(context):
    print(f"=====context: {context}")
    print("=====teacher_math_handoff=====")


def teacher_chinese_handoff(context):
    print(f"=====context: {context}")
    print("=====teacher_chinese_handoff=====")


# 创建一个handoff，指定一个agent
handoff_teacher_math = handoff(
    teacher_math,
    on_handoff=teacher_math_handoff   # 当agent被调用时会执行的函数
)


handoff_teacher_chinese = handoff(
    teacher_chinese,
    on_handoff=teacher_chinese_handoff
)


# 创建一个父agent，下面有两个子agent，父agent可以把任务派发给子agent
teacher_director = Agent(
    name="director",
    instructions="你是一位高中教学主任，你的工作是选择合适的老师解答学生遇到的学习问题，并协调各个老师的工作",
    model=MODEL,
    handoffs=[handoff_teacher_math, handoff_teacher_chinese],   # 指定若干个子agent, 可以是agent对象也可以是handoff对象
)


async def main():
    result = await Runner.run(teacher_director, input="你好，我需要制定一个详细的高中三年数学的学习计划")
    print(result.final_output)


if __name__ == "__main__":
    asyncio.run(main())