Repository navigation
Expand file tree
/
Copy pathtest_gemini_service.py
More file actions
178 lines (144 loc) · 6.83 KB
/
Copy pathtest_gemini_service.py
File metadata and controls
178 lines (144 loc) · 6.83 KB
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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
#!/usr/bin/env python3
"""
Test client for Gemini Hypha Service
This script connects to the Hypha server and tests the Gemini agent service
by sending queries and processing the streaming responses.
"""
import asyncio
import sys
from hypha_rpc import connect_to_server
# Configuration
HYPHA_SERVER_URL = 'https://hypha.aicell.io'
WORKSPACE = 'ws-user-github|478667'
TOKEN = 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2FtdW4uYWkvIiwic3ViIjoic3BhcmtseS1waHJhc2UtNzA2ODE3ODYiLCJhdWQiOiJodHRwczovL2FtdW4tYWkuZXUuYXV0aDAuY29tL2FwaS92Mi8iLCJpYXQiOjE3NTE1MTgwNDYsImV4cCI6MTc4NzUxODA0Niwic2NvcGUiOiJ3czp3cy11c2VyLWdpdGh1Ynw0Nzg2Njcjcncgd2lkOndzLXVzZXItZ2l0aHVifDQ3ODY2NyIsImd0eSI6ImNsaWVudC1jcmVkZW50aWFscyIsImh0dHBzOi8vYW11bi5haS9yb2xlcyI6WyJhZG1pbiJdLCJodHRwczovL2FtdW4uYWkvZW1haWwiOiJvZXdheTAwN0BnbWFpbC5jb20ifQ.rwAm1tkBGuNOuvV2OJ-6VtUVuTgMCqUNnxdJkI-q4Tw'
SERVICE_ID = 'gemini-agent'
async def test_gemini_service():
"""Test the Gemini service by sending queries and processing responses."""
print("Connecting to Hypha server...")
try:
# Connect to Hypha server
server = await connect_to_server({
'server_url': HYPHA_SERVER_URL,
'workspace': WORKSPACE,
'token': TOKEN
})
print(f"Connected to workspace: {server.config.workspace}")
# Get the Gemini service
service_full_id = f"{server.config.workspace}/{SERVICE_ID}"
print(f"Looking for service: {service_full_id}")
# Try to get our specific service
service = None
try:
# First try to find a service by iterating through available services
services = await server.list_services()
for svc in services:
if SERVICE_ID in svc.id and 'gemini-agent' in svc.id:
service = await server.get_service(svc.id)
print(f"✓ Found Gemini service: {service.id}")
break
if not service:
print(f"✗ No {SERVICE_ID} service found in available services")
return
8000
except Exception as e:
print(f"✗ Could not access Gemini service: {e}")
return
# Test queries
test_queries = [
"What is the weather like today?",
"Help me write a Python function to calculate fibonacci numbers",
"Explain how machine learning works in simple terms"
]
for i, query in enumerate(test_queries, 1):
print(f"\n{'='*50}")
print(f"Test {i}: {query}")
print(f"{'='*50}")
try:
# Call the chat service
response_generator = await service.chat(query)
# Process streaming responses
async for response in response_generator:
if isinstance(response, dict):
response_type = response.get('type', 'unknown')
content = response.get('content', '')
timestamp = response.get('timestamp', '')
if response_type == 'status':
print(f"[STATUS] {content}")
elif response_type == 'text':
print(content, end='', flush=True)
elif response_type == 'error':
print(f"\n[ERROR] {content}")
elif response_type == 'final':
print(f"\n[FINAL] Response complete")
else:
print(f"[RESPONSE] {response}")
except Exception as e:
print(f"Error during query {i}: {e}")
import traceback
traceback.print_exc()
print("\n" + "-"*50)
print("\nAll tests completed!")
except Exception as e:
print(f"Failed to connect to Hypha server: {e}")
import traceback
traceback.print_exc()
async def simple_test():
"""Simple test to verify basic connectivity."""
print("Testing basic connectivity...")
try:
# Connect to Hypha server
server = await connect_to_server({
'server_url': HYPHA_SERVER_URL,
'workspace': WORKSPACE,
'token': TOKEN
})
print(f"✓ Connected to workspace: {server.config.workspace}")
# List services
try:
services = await server.list_services()
print(f"✓ Found {len(services)} services:")
for service in services:
print(f" - {service}")
except Exception as e:
print(f"✗ Could not list services: {e}")
# Try to get our specific service
service_full_id = f"{server.config.workspace}/{SERVICE_ID}"
service = None
try:
# First try to find a service by iterating through available services
services = await server.list_services()
for svc in services:
if SERVICE_ID in svc.id and 'gemini-agent' in svc.id:
service = await server.get_service(svc.id)
print(f"✓ Found Gemini service: {service.id}")
break
if not service:
print(f"✗ No {SERVICE_ID} service found in available services")
return
# Test a simple query
print("\nTesting with simple query...")
response_generator = await service.chat("Hello, how are you?")
async for response in response_generator:
if isinstance(response, dict):
response_type = response.get('type', 'unknown')
content = response.get('content', '')
if response_type == 'status':
print(f"[STATUS] {content}")
elif response_type == 'text':
print(content, end='', flush=True)
elif response_type == 'final':
print(f"\n✓ Query completed successfully")
break
elif response_type == 'error':
print(f"\n✗ Error: {content}")
break
except Exception as e:
print(f"✗ Could not access Gemini service: {e}")
except Exception as e:
print(f"✗ Connection failed: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "simple":
asyncio.run(simple_test())
else:
asyncio.run(test_gemini_service())