76 lines
1.8 KiB
Python
76 lines
1.8 KiB
Python
import asyncio
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
from contextlib import aclosing
|
|
|
|
from voice_assistant.agent.oci import OCIAgentAdapter
|
|
|
|
|
|
def get_compartment_id() -> str:
|
|
request = urllib.request.Request(
|
|
"http://169.254.169.254/opc/v2/instance/",
|
|
headers={"Authorization": "Bearer Oracle"},
|
|
)
|
|
|
|
with urllib.request.urlopen(
|
|
request,
|
|
timeout=5,
|
|
) as response:
|
|
metadata = json.load(response)
|
|
|
|
return metadata["compartmentId"]
|
|
|
|
|
|
async def main() -> None:
|
|
agent = OCIAgentAdapter(
|
|
region="sa-vinhedo-1",
|
|
compartment_id=get_compartment_id(),
|
|
model_id="openai.gpt-4o-mini",
|
|
temperature=0.3,
|
|
max_completion_tokens=128,
|
|
timeout_seconds=30.0,
|
|
)
|
|
|
|
started = time.perf_counter()
|
|
first_delta_at = None
|
|
parts = []
|
|
|
|
print("OCI AGENT STREAMING TEST")
|
|
print("-" * 40)
|
|
|
|
async with aclosing(
|
|
agent.stream(
|
|
session_id="smoke-session",
|
|
turn_id=1,
|
|
text=(
|
|
"Explique em duas frases curtas "
|
|
"como funciona um assistente de voz."
|
|
),
|
|
)
|
|
) as stream:
|
|
async for delta in stream:
|
|
if first_delta_at is None:
|
|
first_delta_at = time.perf_counter() - started
|
|
|
|
parts.append(delta)
|
|
print(delta, end="", flush=True)
|
|
|
|
elapsed = time.perf_counter() - started
|
|
response = "".join(parts)
|
|
|
|
if not response.strip():
|
|
raise RuntimeError("OCI Agent returned an empty response.")
|
|
|
|
print("\n")
|
|
print("-" * 40)
|
|
print(f"First text delta: {first_delta_at:.3f}s")
|
|
print(f"Total time: {elapsed:.3f}s")
|
|
print(f"Text chunks: {len(parts)}")
|
|
print("OCI Agent streaming: SUCCESS")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|
|
|