如何做网站站内搜索代码网上营销新观察网

引言

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

完整代码

import asyncio
import random
from typing import Anyfrom pydantic import BaseModelfrom agents import Agent, AgentHooks, RunContextWrapper, Runner, Tool, function_toolclass CustomAgentHooks(AgentHooks):def __init__(self, display_name: str):self.event_counter = 0self.display_name = display_nameasync def on_start(self, context: RunContextWrapper, agent: Agent) -> None:self.event_counter += 1print(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 += 1print(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 += 1print(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 += 1print(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 += 1print(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 * 2class FinalResult(BaseModel):number: intmultiply_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 = 0self.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)

运行效果

执行程序后，我们可以看到完整的代理生命周期：

1. Start Agent启动并生成随机数
2. 根据数字的奇偶性决定是否交接
3. Multiply Agent接收并处理数字
4. 最终输出处理结果

最佳实践

1. 使用钩子机制监控代理行为
2. 合理划分代理职责
3. 实现清晰的代理间通信
4. 使用异步编程提高效率
5. 做好错误处理和日志记录

总结

通过OpenAI Agents框架，我们可以轻松实现智能代理的生命周期管理和多代理协作。这种方式不仅提高了代码的可维护性，也为构建复杂的AI应用提供了坚实的基础。