אחד המאפיינים החשובים והחדשים של מערכות אג'נטיות הוא השימוש ב Streaming. הייחוד של מודלי שפה גדולים הוא שהם מייצרים המון פלט ודי לאט. במערכת אינטרקטיבית זה אומר שמשתמשים צריכים לחכות המון זמן עד שיכולים לכתוב את ההודעה הבאה ולהמשיך את השיחה. אפילו בתהליכים אסינכרוניים זה אומר שלמשתמשים קשה להבין איפה בדיוק אנחנו בתהליך.

הפיתרון של ספקי המודלים היה ליצור APIs שתומכים בהזרמת מידע ולשלוח את הטקסט מהמודל מיד איך שהוא נוצר, מילה אחרי מילה. התוצאה היא חווית ה Streaming שאנחנו מכירים מאתרי ה Chat - אנחנו אומנם מחכים שהמודל יסיים לכתוב את כל הפלט, אבל לפחות במהלך ההמתנה אנחנו יכולים לראות את התוצאה נבנית מול העיניים.

סוכנים שאנחנו בונים לרוב יצטרכו לתת חווית משתמש דומה ולכן כל ספריית עבודה עם סוכנים חייבת לתת פיתרון גם להזרמת המידע. בפוסט זה נראה את הפיתרון של OpenAI Agents SDK.

1. לולאת הזרמת טקסט פשוטה משורת הפקודה

בשביל בדיקות ומשחקים נתנו לנו ב Agents SDK לולאת הזרמת טקסט פשוטה שאפשר להריץ משורת הפקודה. זה הקוד:

import asyncio
from agents import Agent, run_demo_loop

async def main() -> None:
    agent = Agent(name="Assistant", instructions="You are a helpful assistant.")
    await run_demo_loop(agent)

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

לולאה זו משמשת לבדיקת סוכן שעובד בתוך Session ומבצעת:

1. קריאת הודעה מהמשתמש.
2. שמירת ההודעה בזיכרון והעברתה לסוכן.
3. הצגת הפלט מהסוכן ב Streaming, מה שמוכן מוצג.
4. קריאת הודעה נוספת והמשך שיחה, כלומר שליחתה לסוכן בצירוף כל ההודעות הקודמות.

נסו להפעיל את הלולאה אצלכם ולדבר עם סוכן של OpenAI או ספק אחר דרך LiteLLM.

2. הזרמת מידע מקוד שלנו

כמובן שבקוד שלנו אנחנו נעדיף לשלוט בלולאה, בהודעות וב Sessions וגם לא נרצה שהפלט תמיד יוצג למסך ולכן נשתמש בלולאה הדמו רק לבדיקות ובקוד אמיתי נשתמש בפונקציה Runner.run_streamed כדי להריץ סוכן ולקבל ממנו הודעות בהזרמה.

כל הודעה שמתקבלת מהסוכן מגיעה עם type. סוג ההודעה הבסיסי הוא raw_response_event ואלה ההודעות שמעבירות את המילים שהמודל מייצר. סוגי הודעות נוספים כוללים דיווח מתי המודל רוצה להפעיל כלי, מתי המודל סיים הודעה שלמה ומתי קיבלנו תוצאה של כלי. הלולאה הבאה מדפיסה את כל ההודעות:

import asyncio
import random
from agents import Agent, ItemHelpers, Runner, function_tool
from openai.types.responses import ResponseTextDeltaEvent

async def main():
    agent = Agent(
        name="Joker",
        instructions="Tell me a joke",
    )

    result = Runner.run_streamed(
        agent,
        input="Hello",
    )
    print("=== Run starting ===")

    async for event in result.stream_events():
        # We'll ignore the raw responses event deltas
        if event.type == "raw_response_event":
            if isinstance(event.data, ResponseTextDeltaEvent):
                print(f"[Token]: {event.data.delta}")
            else:
                print(event)

        # When the agent updates, print that
        elif event.type == "agent_updated_stream_event":
            print(f"Agent updated: {event.new_agent.name}")
            continue

        # When items are generated, print them
        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:
                print(f" -- Other Event: {event}")

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


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

הרצת התוכנית מדפיסה את הפלט הבא:

/Users/ynonp/work/projects/ai/leanagentssdk/.venv/bin/python /Users/ynonp/work/projects/ai/leanagentssdk/streaming.py 
=== Run starting ===
Agent updated: Joker
RawResponsesStreamEvent(data=ResponseCreatedEvent(response=Response(id='resp_6895b0fd4464819abe44eb9e456d2d960754fc8434302a7d', created_at=1754640637.0, error=None, incomplete_details=None, instructions='Tell me a joke', metadata={}, model='gpt-4o-2024-08-06', object='response', output=[], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, max_output_tokens=None, max_tool_calls=None, previous_response_id=None, prompt=None, reasoning=Reasoning(effort=None, generate_summary=None, summary=None), service_tier='auto', status='in_progress', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=None, user=None, prompt_cache_key=None, safety_identifier=None, store=True), sequence_number=0, type='response.created'), type='raw_response_event')
RawResponsesStreamEvent(data=ResponseInProgressEvent(response=Response(id='resp_6895b0fd4464819abe44eb9e456d2d960754fc8434302a7d', created_at=1754640637.0, error=None, incomplete_details=None, instructions='Tell me a joke', metadata={}, model='gpt-4o-2024-08-06', object='response', output=[], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, max_output_tokens=None, max_tool_calls=None, previous_response_id=None, prompt=None, reasoning=Reasoning(effort=None, generate_summary=None, summary=None), service_tier='auto', status='in_progress', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=None, user=None, prompt_cache_key=None, safety_identifier=None, store=True), sequence_number=1, type='response.in_progress'), type='raw_response_event')
RawResponsesStreamEvent(data=ResponseOutputItemAddedEvent(item=ResponseOutputMessage(id='msg_6895b0fda330819a9278cd18b1f7a04b0754fc8434302a7d', content=[], role='assistant', status='in_progress', type='message'), output_index=0, sequence_number=2, type='response.output_item.added'), type='raw_response_event')
RawResponsesStreamEvent(data=ResponseContentPartAddedEvent(content_index=0, item_id='msg_6895b0fda330819a9278cd18b1f7a04b0754fc8434302a7d', output_index=0, part=ResponseOutputText(annotations=[], text='', type='output_text', logprobs=[]), sequence_number=3, type='response.content_part.added'), type='raw_response_event')
[Token]: Hi
[Token]:  there
[Token]: !
[Token]:  Ready
[Token]:  for
[Token]:  a
[Token]:  joke
[Token]: ?
RawResponsesStreamEvent(data=ResponseTextDoneEvent(content_index=0, item_id='msg_6895b0fda330819a9278cd18b1f7a04b0754fc8434302a7d', logprobs=[], output_index=0, sequence_number=12, text='Hi there! Ready for a joke?', type='response.output_text.done'), type='raw_response_event')
RawResponsesStreamEvent(data=ResponseContentPartDoneEvent(content_index=0, item_id='msg_6895b0fda330819a9278cd18b1f7a04b0754fc8434302a7d', output_index=0, part=ResponseOutputText(annotations=[], text='Hi there! Ready for a joke?', type='output_text', logprobs=[]), sequence_number=13, type='response.content_part.done'), type='raw_response_event')
RawResponsesStreamEvent(data=ResponseOutputItemDoneEvent(item=ResponseOutputMessage(id='msg_6895b0fda330819a9278cd18b1f7a04b0754fc8434302a7d', content=[ResponseOutputText(annotations=[], text='Hi there! Ready for a joke?', type='output_text', logprobs=[])], role='assistant', status='completed', type='message'), output_index=0, sequence_number=14, type='response.output_item.done'), type='raw_response_event')
RawResponsesStreamEvent(data=ResponseCompletedEvent(response=Response(id='resp_6895b0fd4464819abe44eb9e456d2d960754fc8434302a7d', created_at=1754640637.0, error=None, incomplete_details=None, instructions='Tell me a joke', metadata={}, model='gpt-4o-2024-08-06', object='response', output=[ResponseOutputMessage(id='msg_6895b0fda330819a9278cd18b1f7a04b0754fc8434302a7d', content=[ResponseOutputText(annotations=[], text='Hi there! Ready for a joke?', type='output_text', logprobs=[])], role='assistant', status='completed', type='message')], parallel_tool_calls=True, temperature=1.0, tool_choice='auto', tools=[], top_p=1.0, background=False, max_output_tokens=None, max_tool_calls=None, previous_response_id=None, prompt=None, reasoning=Reasoning(effort=None, generate_summary=None, summary=None), service_tier='default', status='completed', text=ResponseTextConfig(format=ResponseFormatText(type='text'), verbosity='medium'), top_logprobs=0, truncation='disabled', usage=ResponseUsage(input_tokens=16, input_tokens_details=InputTokensDetails(cached_tokens=0), output_tokens=9, output_tokens_details=OutputTokensDetails(reasoning_tokens=0), total_tokens=25), user=None, prompt_cache_key=None, safety_identifier=None, store=True), sequence_number=15, type='response.completed'), type='raw_response_event')
-- Message output:
 Hi there! Ready for a joke?
=== Run complete ===

Process finished with exit code 0

אפשר לראות את כל הטוקנים מודפסים בזמן שהם מיוצרים, ואחריהם הודעת סיום עם ההודעה המלאה. בתוכניות אמיתיות נרצה לטפל בצורה שונה בכל סוג הודעה, לדוגמה:

1. הודעות raw_response_event יישלחו לגולש בזמן שהוא ממתין לתשובת המודל כדי להראות את התשובה בזמן שהיא נוצרת.

2. הודעות tool_call_item יציגו למשתמש סימן מיוחד בממשק שאומר שהמודל עכשיו מפעיל כלי, למשל מחפש באינטרנט.

3. הודעות message_output_item יעדכנו את ההודעה שמופיעה בחלון השיחה עם הטקסט המלא של ההודעה. אפשר גם להשתמש באירוע זה כדי לשמור ב DB את ההודעה.

3. עכשיו אתם

הוסיפו לתוכנית שלנו Tool כלשהו, למשל פונקציה שמחזירה בצורה אקראית נושא לבדיחה. וודאו שהמודל מפעיל את ה Tool ושימו לב להודעות שמתקבלות.

בפוסטים הבאים נראה איך להזרים את תשובות המודל לממשקי ווב, לבוט בטלגרם וגם ללקוח MCP.