Implementing a streaming response requires utilizing the ZipAgent providedrun_streammethods and event handling mechanisms:
Basic realization steps:
1. Callsrun_streamalternativerunmethod to get the event stream
2. Iterate over different types of event objects
3. TargetingANSWER_DELTAEvent real-time output content
stream = Runner.run_stream(agent, "解释量子计算原理")
for event in stream:
if event.type == StreamEventType.ANSWER_DELTA:
print(event.content, end="", flush=True)
elif event.type == StreamEventType.TOOL_CALL:
print(f"n调用工具: {event.tool_name}")
Advanced Application Tips:
- transferring entity
flush=Trueparameter ensures timely output of individual characters - Enhancements such as colorful progress indicators can be achieved by combining ANSI escapes.
- For complex scenarios, event streams can be cached for post-processing
The mechanism not only realizes character-level real-time output effect, but also gives clear hints when the tool is invoked, significantly improving the naturalness and transparency of human-computer interaction, which is a key feature for building professional-grade dialogue experience.
This answer comes from the articleZipAgent: a lightweight Python framework for building exclusive AI assistants in 5 minutesThe
