feat: implement AI-orchestrated result extraction and service connectivity tests

2026-08-27 17:45:07 +05:30
parent 59c1372fe3
commit 3d2596ced1
11 changed files with 546 additions and 123 deletions
@@ -0,0 +1,87 @@
+import asyncio
+import os
+import sys
+from dotenv import load_dotenv
+
+if sys.stdout.encoding != 'utf-8':
+    try:
+        sys.stdout.reconfigure(encoding='utf-8')
+    except Exception:
+        pass
+
+load_dotenv()
+
+async def test_sarvam_v2():
+    api_key = os.getenv("SARVAM_API_KEY")
+    import httpx2
+    client = httpx2.AsyncClient(timeout=15.0)
+    headers = {
+        "api-subscription-key": api_key,
+        "Content-Type": "application/json",
+    }
+    payload = {
+        "inputs": ["வணக்கம், உங்கள் உடல்நலம் எப்படி உள்ளது?"],
+        "target_language_code": "ta-IN",
+        "speaker": "kavya",
+        "model": "bulbul:v2",
+    }
+    res = await client.post(
+        "https://api.sarvam.ai/text-to-speech",
+        json=payload,
+        headers=headers,
+    )
+    if res.status_code == 200:
+        audios = res.json().get("audios", [])
+        print(f"[OK] Sarvam TTS bulbul:v2 WORKS! Generated {len(audios)} audio clips.")
+        return True
+    else:
+        print(f"[FAIL] Sarvam error {res.status_code}: {res.text}")
+        return False
+
+async def test_gemini_models():
+    api_key = os.getenv("GEMINI_API_KEY")
+    import google.generativeai as genai
+    genai.configure(api_key=api_key)
+    for model_name in ["gemini-3.6-flash", "gemini-flash-latest", "gemini-2.5-flash-lite", "gemma-4-26b-a4b-it"]:
+        try:
+            m = genai.GenerativeModel(model_name)
+            res = await m.generate_content_async("Hello")
+            print(f"[OK] Gemini '{model_name}' WORKS! Response: {res.text.strip()}")
+            return model_name
+        except Exception as e:
+            print(f"[X] Gemini '{model_name}' failed: {e}")
+    return None
+
+async def test_anthropic_models():
+    api_key = os.getenv("ANTHROPIC_API_KEY")
+    from anthropic import AsyncAnthropic
+    client = AsyncAnthropic(api_key=api_key)
+    for model_name in [
+        "claude-3-7-sonnet-20250219",
+        "claude-3-5-sonnet-20240620",
+        "claude-3-sonnet-20240229",
+        "claude-2.1",
+        "claude-instant-1.2"
+    ]:
+        try:
+            res = await client.messages.create(
+                model=model_name,
+                max_tokens=20,
+                messages=[{"role": "user", "content": "Hi"}],
+            )
+            print(f"[OK] Anthropic '{model_name}' WORKS! Response: {res.content[0].text.strip()}")
+            return model_name
+        except Exception as e:
+            print(f"[X] Anthropic '{model_name}' failed: {e}")
+    return None
+
+async def main():
+    print("--- Testing Sarvam v2 ---")
+    await test_sarvam_v2()
+    print("\n--- Testing Gemini Models ---")
+    await test_gemini_models()
+    print("\n--- Testing Anthropic Models ---")
+    await test_anthropic_models()
+
+if __name__ == "__main__":
+    asyncio.run(main())