目录

使用OpenAI-Agents实现流式输出

使用OpenAI Agents实现流式输出

引言

在AI应用开发中,流式输出(Stream Output)是提升用户体验的重要特性。本文将介绍如何使用OpenAI Agents框架实现一个能够实时生成笑话的智能助手,并通过流式输出实现交互式的响应效果。

完整代码

import asyncio
import random

from agents import Agent, ItemHelpers, Runner, function_tool


@function_tool
def how_many_jokes() -> int:
    return random.randint(1, 10)

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():
    agent = Agent(
        name="Joker",
        instructions="First call the `how_many_jokes` tool, then tell that many jokes.",
        tools=[how_many_jokes],
        model="glm-4"
    )

    result = Runner.run_streamed(
        agent,
        input="Hello",
        run_config=run_config
    )
    print("=== Run starting ===")
    async for event in result.stream_events():
        # We'll ignore the raw responses event deltas
        if event.type == "raw_response_event":
            continue
        elif event.type == "agent_updated_stream_event":
            print(f"Agent updated: {event.new_agent.name}")
            continue
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print("-- Tool was called")
            elif event.item.type == "tool_call_output_item":
                print(f"-- Tool output: {event.item.output}")
            elif event.item.type == "message_output_item":
                print(f"-- Message output:\n {ItemHelpers.text_message_output(event.item)}")
            else:
                pass  # Ignore other event types

    print("=== Run complete ===")


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

"""
$python .My_test\stream_itemstest.py  
=== Run starting ===
Agent updated: Joker
Agent updated: Joker
-- Tool was called
-- Message output:
 Here are 9 jokes for you:
1. Why don't scientists trust atoms? Because they make up everything!
2. Why did the scarecrow win an award? Because he was outstanding in his field!
3. Why did the math book look sad? Because it had too many problems!
4. Why did the cookie go to the doctor? Because it was feeling crummy!
5. Why did the bicycle fall over? Because it was two-tired!
6. Why did the cat sit on the computer? Because it wanted to keep an eye on the mouse!
7. Why did the teacher wear sunglasses? Because her students were so bright!
8. Why did the chicken cross the playground? To get to the other slide!
9. Why did the bookworm go to the gym? Because it heard the exercise was good for the mind!     
=== Run complete ===
OPENAI_API_KEY is not set, skipping trace export
"""

核心功能实现

1. 自定义工具函数

@function_tool
def how_many_jokes() -> int:
    return random.randint(1, 10)

通过装饰器 @function_tool ,我们创建了一个简单的工具函数,用于随机决定生成笑话的数量。

2. 智能代理配置

agent = Agent(
    name="Joker",
    instructions="First call the `how_many_jokes` tool, then tell that many jokes.",
    tools=[how_many_jokes],
    model="glm-4"
)

技术特点

1. 异步流式处理

  • 使用 asyncio 实现异步操作
  • 通过 stream_events() 获取实时事件
  • 支持多种事件类型的处理

2. 事件类型处理

主要处理三种类型的事件:

  1. raw_response_event:原始响应事件
  2. agent_updated_stream_event:代理更新事件
  3. run_item_stream_event:运行项目事件

3. 实时反馈机制

async for event in result.stream_events():
    if event.type == "run_item_stream_event":
        if event.item.type == "tool_call_item":
            print("-- Tool was called")
        elif event.item.type == "message_output_item":
            print(f"-- Message output:\n {ItemHelpers.text_message_output(event.item)}")

应用场景

  1. 娱乐应用

    • 智能笑话生成器
    • 互动式故事创作
    • 实时对话系统
  2. 教育领域

    • 趣味学习助手
    • 互动式教学工具
    • 学生参与度提升
  3. 客户服务

    • 实时响应系统
    • 智能客服助手
    • 用户体验优化

实现效果

运行效果展示:

=== Run starting ===
Agent updated: Joker
-- Tool was called
-- Message output:
 Here are 9 jokes for you:
1. Why don't scientists trust atoms? ...
2. Why did the scarecrow win an award? ...
[更多笑话...]
=== Run complete ===

最佳实践

  1. 事件处理

    • 合理分类事件类型
    • 优雅处理异常情况
    • 清晰的日志输出
  2. 性能优化

    • 异步处理提升效率
    • 合理的事件过滤
    • 资源使用优化
  3. 用户体验

    • 实时反馈
    • 清晰的输出格式
    • 友好的交互方式

总结

通过OpenAI Agents框架实现的流式输出功能,不仅提供了更好的用户体验,还为AI应用开发提供了新的可能性。这种方法特别适合需要实时反馈的应用场景。

注意事项

由于智谱AI的API返回的用量统计格式与OpenAI的格式不同导致的。错误显示 CompletionUsage 对象没有 input_tokens 属性。

智谱的usage token为

https://i-blog.csdnimg.cn/direct/d94f2423ea61433cadc3c1e2d7673f67.png

修改成

https://i-blog.csdnimg.cn/direct/94903d8ffeb34697aed7896dc2821fe4.png