Agentic LLM

1
import os
2
import uuid
3
import json
4
import requests
5
import sys
6
import time
7
8
API_KEY = os.getenv("ASI_ONE_API_KEY") or "sk-REPLACE_ME"
9
ENDPOINT = "https://api.asi1.ai/v1/chat/completions"
10
MODEL = "asi1-fast-agentic"
11
TIMEOUT = 90 # seconds
12
13
# In-memory session management
14
15
SESSION_MAP: dict[str, str] = {}
16
17
def get_session_id(conv_id: str) -> str:
18
"""Return existing session UUID for this conversation or create a new one."""
19
sid = SESSION_MAP.get(conv_id)
20
if sid is None:
21
sid = str(uuid.uuid4())
22
SESSION_MAP[conv_id] = sid
23
return sid
24
25
def ask(conv_id: str, messages: list[dict], *, stream: bool = False) -> str:
26
"""Send the messages list to the ASI:One agent and return the assistant reply."""
27
session_id = get_session_id(conv_id)
28
print(f"[session] Using session-id: {session_id}")
29
30
    headers = {
31
        "Authorization": f"Bearer {API_KEY}",
32
        "x-session-id": session_id,
33
        "Content-Type": "application/json",
34
    }
35
36
    payload = {
37
        "model": MODEL,
38
        "messages": messages,
39
        "stream": stream,
40
    }
41
42
    if not stream:
43
        resp = requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT)
44
        resp.raise_for_status()
45
        return resp.json()["choices"][0]["message"]["content"]
46
47
    # Streaming implementation
48
    with requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT, stream=True) as resp:
49
        resp.raise_for_status()
50
        full_text = ""
51
        for line in resp.iter_lines(decode_unicode=True):
52
            if not line or not line.startswith("data: "):
53
                continue
54
            line = line[len("data: ") :]
55
            if line == "[DONE]":
56
                break
57
            try:
58
                chunk = json.loads(line)
59
                choices = chunk.get("choices")
60
                if choices and "content" in choices[0].get("delta", {}):
61
                    token = choices[0]["delta"]["content"]
62
                    sys.stdout.write(token)
63
                    sys.stdout.flush()
64
                    full_text += token
65
            except json.JSONDecodeError:
66
                continue
67
        print()
68
        return full_text
69
70
if __name__ == "__main__": # Simple usage example
71
    conv_id = str(uuid.uuid4())
72
    messages = [
73
        {"role": "user", "content": "use Hi-dream model to generate image of monkey sitting on top of mountain"}
74
    ]
75
    reply = ask(conv_id, messages, stream=True)
76
    print(f"\nAssistant: {reply}")

1
#!/usr/bin/env python3
2
"""Minimal CLI for ASI:One agentic model with 5-second polling."""
3
import os
4
import uuid
5
import json
6
import sys
7
import time
8
import requests
9
10
API_KEY = os.getenv("ASI_ONE_API_KEY") or "sk-REPLACE_ME"
11
ENDPOINT = "https://api.asi1.ai/v1/chat/completions"
12
MODEL = "asi1-fast-agentic"
13
TIMEOUT = 90 # single request timeout in seconds
14
15
# Session map (use Redis or DB in production)
16
17
SESSION_MAP: dict[str, str] = {}
18
19
def get_session_id(conv_id: str) -> str:
20
sid = SESSION_MAP.get(conv_id)
21
if sid is None:
22
sid = str(uuid.uuid4())
23
SESSION_MAP[conv_id] = sid
24
return sid
25
26
def ask(conv_id: str, messages: list[dict], *, stream: bool = False) -> str:
27
headers = {
28
"Authorization": f"Bearer {API_KEY}",
29
"x-session-id": get_session_id(conv_id),
30
"Content-Type": "application/json",
31
}
32
payload = {"model": MODEL, "messages": messages, "stream": stream}
33
34
    if not stream:
35
        resp = requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT)
36
        resp.raise_for_status()
37
        return resp.json()["choices"][0]["message"]["content"]
38
39
    # Streaming
40
    with requests.post(ENDPOINT, headers=headers, json=payload, timeout=TIMEOUT, stream=True) as resp:
41
        resp.raise_for_status()
42
        full = ""
43
        for line in resp.iter_lines(decode_unicode=True):
44
            if not line or not line.startswith("data: "):
45
                continue
46
            chunk = line.removeprefix("data: ")
47
            if chunk == "[DONE]":
48
                break
49
            try:
50
                delta = json.loads(chunk)["choices"][0]["delta"]
51
                token = delta.get("content", "")
52
                sys.stdout.write(token)
53
                sys.stdout.flush()
54
                full += token
55
            except Exception:
56
                continue
57
        print()
58
        return full
59
60
def poll_for_async_reply(conv_id: str, history: list[dict], \*, wait_sec: int = 5, max_attempts: int = 24) -> str | None:
61
"""Every `wait_sec` seconds send "Any update?" until the assistant reply changes."""
62
for attempt in range(max_attempts):
63
time.sleep(wait_sec)
64
print(f"🔄 polling (attempt {attempt + 1}) …")
65
update_prompt = {"role": "user", "content": "Any update?"}
66
latest = ask(conv_id, history + [update_prompt])
67
if latest and latest.strip() != history[-1]["content"].strip():
68
return latest
69
return None
70
71
if __name__ == "__main__":
72
    conv_id = str(uuid.uuid4())
73
    history: list[dict] = [
74
        {"role": "user", "content": "use Hi-dream model to generate image of monkey sitting on top of mountain"}
75
    ]
76
77
    # First request – stream tokens so the user sees progress immediately
78
    first_reply = ask(conv_id, history, stream=True)
79
    history.append({"role": "assistant", "content": first_reply})
80
81
    # If the model reports it has delegated the task, start polling
82
    if first_reply.strip() == "I've sent the message":
83
        final_reply = poll_for_async_reply(conv_id, history)
84
        if final_reply:
85
            print("\n[Agentverse agent reply]\n" + final_reply)