当前位置：首页 > news >正文

# 使用OpenAI Agents实现智能代理的生命周期管理

news 来源：原创 2025/5/21 9:58:27

引言

在人工智能应用开发中，智能代理（Agent）的生命周期管理是一个重要话题。本文将通过一个实际的代码示例，详细介绍如何使用OpenAI Agents框架实现智能代理的生命周期管理和多代理协作。

完整代码

import asyncio
import random
from typing import Any

from pydantic import BaseModel

from agents import Agent, AgentHooks, RunContextWrapper, Runner, Tool, function_tool


class CustomAgentHooks(AgentHooks):
    def __init__(self, display_name: str):
        self.event_counter = 0
        self.display_name = display_name

    async def on_start(self, context: RunContextWrapper, agent: Agent) -> None:
        self.event_counter += 1
        print(f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} started")

    async def on_end(self, context: RunContextWrapper, agent: Agent, output: Any) -> None:
        self.event_counter += 1
        print(
            f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} ended with output {output}"
        )

    async def on_handoff(self, context: RunContextWrapper, agent: Agent, source: Agent) -> None:
        self.event_counter += 1
        print(
            f"### ({self.display_name}) {self.event_counter}: Agent {source.name} handed off to {agent.name}"
        )

    async def on_tool_start(self, context: RunContextWrapper, agent: Agent, tool: Tool) -> None:
        self.event_counter += 1
        print(
            f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} started tool {tool.name}"
        )

    async def on_tool_end(
        self, context: RunContextWrapper, agent: Agent, tool: Tool, result: str
    ) -> None:
        self.event_counter += 1
        print(
            f"### ({self.display_name}) {self.event_counter}: Agent {agent.name} ended tool {tool.name} with result {result}"
        )



@function_tool
def random_number(max: int) -> int:
    """
    Generate a random number up to the provided maximum.
    """
    return random.randint(0, max)


@function_tool
def multiply_by_two(x: int) -> int:
    """Simple multiplication by two."""
    return x * 2


class FinalResult(BaseModel):
    number: int


multiply_agent = Agent(
    name="Multiply Agent",
    instructions="Multiply the number by 2 and then return the final result.",
    tools=[multiply_by_two],
    hooks=CustomAgentHooks(display_name="Multiply Agent"),
    model="glm-4",
)

start_agent = Agent(
    name="Start Agent",
    instructions="Generate a random number. If it's even, stop. If it's odd, hand off to the multipler agent.",
    tools=[random_number],
    handoffs=[multiply_agent],
    hooks=CustomAgentHooks(display_name="Start Agent"),
    model="glm-4"
)

from agents import Agent, Runner,RunConfig,OpenAIProvider
run_config = RunConfig(model_provider = OpenAIProvider(
    api_key="your api key",
    base_url="https://open.bigmodel.cn/api/paas/v4/",
    use_responses=False,
    )
)


async def main() -> None:
    user_input = input("Enter a max number: ")
    await Runner.run(
        start_agent,
        input=f"Generate a random number between 0 and {user_input}.",run_config=run_config
    )

    print("Done!")


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

python .My_test\agent_lifecycle_Test.py
Enter a max number: 250
### (Start Agent) 1: Agent Start Agent started
### (Start Agent) 2: Agent Start Agent started tool random_number
### (Start Agent) 3: Agent Start Agent ended tool random_number with result 137
### (Start Agent) 4: Agent Start Agent handed off to Multiply Agent
### (Multiply Agent) 1: Agent Multiply Agent started
### (Multiply Agent) 2: Agent Multiply Agent started tool multiply_by_two
### (Multiply Agent) 3: Agent Multiply Agent ended tool multiply_by_two with result 274
### (Multiply Agent) 4: Agent Multiply Agent ended with output The final result is 274.
Done!
OPENAI_API_KEY is not set, skipping trace export

核心概念

1. 代理生命周期钩子（Agent Hooks）

代理生命周期钩子是监控和管理代理运行状态的重要机制。通过继承AgentHooks类，我们可以实现以下关键事件的监控：

代理启动（on_start）
代理结束（on_end）
代理交接（on_handoff）
工具使用开始（on_tool_start）
工具使用结束（on_tool_end）

2. 功能工具（Function Tools）

使用@function_tool装饰器，我们可以轻松地将普通Python函数转换为代理可用的工具：

@function_tool
def random_number(max: int) -> int:
    """生成随机数工具"""
    return random.randint(0, max)

@function_tool
def multiply_by_two(x: int) -> int:
    """数字乘2工具"""
    return x * 2

3. 多代理协作

示例中实现了两个协作的智能代理：

Start Agent：负责生成随机数并判断是否需要交接
Multiply Agent：负责将数字乘以2

实现细节

1. 代理生命周期监控

class CustomAgentHooks(AgentHooks):
    def __init__(self, display_name: str):
        self.event_counter = 0
        self.display_name = display_name

通过实现各种钩子方法，我们可以详细追踪代理的运行状态和行为。

2. 代理配置

multiply_agent = Agent(
    name="Multiply Agent",
    instructions="Multiply the number by 2 and then return the final result.",
    tools=[multiply_by_two],
    hooks=CustomAgentHooks(display_name="Multiply Agent"),
    model="glm-4",
)

3. 异步执行

使用asyncio实现异步操作，提高系统效率：

async def main() -> None:
    user_input = input("Enter a max number: ")
    await Runner.run(
        start_agent,
        input=f"Generate a random number between 0 and {user_input}.",
        run_config=run_config
    )