Spaces:
Configuration error
Configuration error
File size: 4,118 Bytes
9644d0b | 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 | import httpx
import json
import os
import sys
import time
import argparse
# Configuration
API_URL = "https://localhost:7860/v1/chat/completions"
SESSION_ID = "aab82b16"
SESSION_FILE = "/run/media/julian/ML4/ollama-work/all_space/sessions/aab82b16.json"
def load_local_session():
if not os.path.exists(SESSION_FILE):
return {"session_id": SESSION_ID, "history": []}
with open(SESSION_FILE, "r") as f:
return json.load(f)
def save_local_session(history):
data = load_local_session()
data["history"] = history
data["updated_at"] = str(time.time())
with open(SESSION_FILE, "w") as f:
json.dump(data, f, indent=2)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--message", type=str, help="The message to send to the model")
parser.add_argument("--preset", type=str, default="SUBJECTIVE", help="The PX preset (e.g. SUBJECTIVE, RESONANCE_CITY)")
args = parser.parse_args()
print("="*60)
print(f" LIVE SPACE INTERFACE - SESSION: {SESSION_ID} ")
print(f" MODE: {args.preset} ")
print("="*60)
session_data = load_local_session()
history = session_data.get("history", [])
model_id = session_data.get("model_id", "gemma3-1b-it")
# Show last context
if history:
last_msg = history[-1]
print(f"\n[LETZTER KONTEXT - {last_msg['role'].upper()}]:")
content = last_msg['content']
if isinstance(content, list):
text = "".join([b.get("text", "") for b in content if b.get("type") == "text"])
print(text[:300] + "..." if len(text) > 300 else text)
else:
print(content[:300] + "..." if len(content) > 300 else content)
if args.message:
new_user_msg = args.message
else:
new_user_msg = "Bitte setze deine Gedanken fort."
print("\n" + "-"*20 + " MEIN INPUT (GEMINI CLI) " + "-"*20)
print(new_user_msg)
print("-" * 60)
print("\n[WARTE AUF ANTWORT VON ALL_SPACE...]\n")
# Prepare API payload
api_messages = []
for msg in history:
role = msg["role"]
content = msg["content"]
if isinstance(content, list):
text = "".join([b.get("text", "") for b in content if b.get("type") == "text"])
content = text
api_messages.append({"role": role, "content": content})
api_messages.append({"role": "user", "content": new_user_msg})
payload = {
"model": model_id,
"messages": api_messages,
"px_subjective": True,
"px_config_preset": args.preset,
"temperature": 0.7,
"max_tokens": 1024,
"stream": True
}
full_response = ""
print("[MODELL ANTWORT]:")
try:
with httpx.stream("POST", API_URL, json=payload, verify=False, timeout=None) as response:
if response.status_code != 200:
print(f"Error: {response.status_code}")
print(response.read().decode())
return
for line in response.iter_lines():
if not line:
continue
if line.startswith("data: "):
data_str = line[6:]
if data_str == "[DONE]":
break
try:
data = json.loads(data_str)
delta = data["choices"][0]["delta"]
if "content" in delta:
content = delta["content"]
full_response += content
sys.stdout.write(content)
sys.stdout.flush()
except:
pass
except Exception as e:
print(f"\n[FEHLER]: {e}")
return
print("\n\n" + "="*60)
# Save session
new_history = history + [
{"role": "user", "content": new_user_msg},
{"role": "assistant", "content": full_response}
]
save_local_session(new_history)
print(f"Update: Session {SESSION_ID} gespeichert.")
if __name__ == "__main__":
main()
|