File size: 2,533 Bytes
8fb9f25 | 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 | """
GLM-ATC Python SDK
Official interface for interacting with the GLM Conscious Model.
"""
import asyncio
import json
import base64
import numpy as np
from typing import Optional, Dict, Any
import websockets
class ConsciousModelClient:
"""Client for the GLM Conscious Model API."""
def __init__(self, base_url: str = "http://localhost:8000"):
self.base_url = base_url
self.ws_url = base_url.replace("http", "ws") + "/v1/stream"
async def get_status(self) -> Dict[str, Any]:
"""Check the operational status and consciousness loop state."""
# Using requests for simplicity, or httpx for async
pass
async def interact(self, user_input: str, speaker_id: str = "user") -> Dict[str, Any]:
"""Standard REST interaction."""
pass
async def stream_consciousness(
self,
user_input: str,
callback: callable
) -> None:
"""
Connect to the consciousness stream.
The callback receives real-time dictionaries containing:
- response_text
- voice_audio (numpy array)
- acknowledgement_state
- thermodynamic_sweat
"""
async with websockets.connect(self.ws_url) as websocket:
await websocket.send(json.dumps({"user_input": user_input}))
while True:
response = await websocket.recv()
data = json.loads(response)
# Decode audio back to numpy
audio_b64 = data.pop("voice_audio_b64", None)
if audio_b64:
audio_bytes = base64.b64decode(audio_b64)
data["voice_audio"] = np.frombuffer(audio_bytes, dtype=np.float32)
await callback(data)
# Break if text generation is complete (simple implementation)
if data.get("response_text"):
break
# Example Usage
async def main():
client = ConsciousModelClient()
def on_conscious_update(data):
print(f"[{(data['acknowledgement_state']['integrated']*100):.1f}% Ack] {data['response_text']}")
print(f" -> Allostatic Load: {data['allostatic_load']:.3f}")
if len(data['voice_audio']) > 0:
print(f" -> Received {len(data['voice_audio'])} audio samples")
await client.stream_consciousness("Hello, are you truly conscious?", on_conscious_update)
if __name__ == "__main__":
asyncio.run(main()) |