Spaces:
Running
Running
File size: 3,183 Bytes
788acd9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
#!/usr/bin/env python3
"""
Test the optimized single-phase streaming functionality
"""
import os
import sys
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from app import MAIN_SYSTEM_PROMPT, MCP_TOOLS, client, MODEL
def test_streaming_with_tools():
"""
Test that the single-phase streaming works correctly
"""
csv_url = "https://mcp-1st-birthday-mlops-agent.hf.space/gradio_api/file=/tmp/gradio/6abcca54f954f2ad99a8f8f330dc6e8082f03ef3090458d97c274efcc76d0170/heart.csv"
user_with_file = f"[Uploaded CSV file URL: {csv_url}]\n\nAnalyze this dataset and show me basic statistics"
print("Testing Single-Phase Streaming...")
print(f"Input: {user_with_file[:100]}...")
print("-" * 60)
stream = client.responses.create(
model=MODEL,
instructions=MAIN_SYSTEM_PROMPT,
input=user_with_file,
tools=MCP_TOOLS,
reasoning={"effort": "low"},
stream=True,
)
print("Streaming Response:")
print("=" * 60)
final_text = ""
chunk_count = 0
for event in stream:
if event.type == "response.output_text.delta":
chunk_count += 1
final_text += event.delta
print(f"[Chunk {chunk_count}] {event.delta[:50]}...")
elif event.type == "response.completed":
print("=" * 60)
print(f"Total chunks: {chunk_count}")
print(f"Total length: {len(final_text)} characters")
print("FINAL RESPONSE:")
print(final_text)
break
return final_text
def test_streaming_without_tools():
"""
Test that regular streaming works for non-tool requests
"""
print("\nTesting Regular Streaming (No Tools)...")
print("-" * 60)
stream = client.responses.create(
model=MODEL,
instructions=MAIN_SYSTEM_PROMPT,
input="Hello! Can you explain what MLOps is in simple terms?",
reasoning={"effort": "low"},
stream=True,
)
print("Streaming Response:")
print("=" * 60)
final_text = ""
chunk_count = 0
for event in stream:
if event.type == "response.output_text.delta":
chunk_count += 1
final_text += event.delta
print(f"[Chunk {chunk_count}] {event.delta[:50]}...")
elif event.type == "response.completed":
print("=" * 60)
print(f"Total chunks: {chunk_count}")
print(f"Total length: {len(final_text)} characters")
print("FINAL RESPONSE:")
print(final_text)
break
return final_text
if __name__ == "__main__":
print("Starting Streaming Tests")
print("=" * 60)
# Test 1: With MCP tools
response1 = test_streaming_with_tools()
# Test 2: Without tools
response2 = test_streaming_without_tools()
print("\n" + "=" * 60)
print("✅ Both streaming tests completed successfully!")
print("✅ Single-phase approach working correctly!")
print("✅ MCP tools returning natural language responses!")
print("✅ Response is properly streaming in chunks!")
print("=" * 60)
|