primeira versão customizada de VAD- Voice Activity Detection

tools/deepagents_adapter_smoke_test.py

@@ -0,0 +1,64 @@
+
+"""
+Smoke test for DeepAgentsAdapter streaming and context.
+"""
+
+import asyncio
+import os
+from contextlib import aclosing
+
+from voice_assistant.agent.deep_agent import DeepAgentsAdapter
+
+
+async def main():
+    compartment_id = os.environ.get("OCI_COMPARTMENT_ID")
+
+    if not compartment_id:
+        raise RuntimeError("OCI_COMPARTMENT_ID is required.")
+
+    adapter = DeepAgentsAdapter(
+        region="sa-vinhedo-1",
+        compartment_id=compartment_id,
+        model_id="openai.gpt-4o-mini",
+        temperature=0.3,
+        max_completion_tokens=256,
+    )
+
+    try:
+        questions = [
+            "Estou planejando visitar o Japão. Lembre desse destino.",
+            "Qual país mencionei? Responda somente com o nome.",
+        ]
+
+        for turn_id, question in enumerate(questions, 1):
+            print(f"\n--- Turn {turn_id} ---")
+            print("User:", question)
+
+            chunks = []
+
+            async with aclosing(
+                adapter.stream(
+                    session_id="agentvoice-adapter-smoke",
+                    turn_id=turn_id,
+                    text=question,
+                )
+            ) as stream:
+                async for chunk in stream:
+                    chunks.append(chunk)
+                    print(chunk, end="", flush=True)
+
+            response = "".join(chunks).strip()
+
+            if not response:
+                raise RuntimeError(
+                    f"Turn {turn_id} returned empty streaming text."
+                )
+
+            print(f"\nChunks: {len(chunks)}")
+
+    finally:
+        await adapter.aclose()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())