import streamlit as st
import os, requests, hmac, re, json, base64
from concurrent.futures import ThreadPoolExecutor, as_completed
st.set_page_config(page_title="Yai", page_icon="✦", layout="wide", initial_sidebar_state="collapsed")
st.markdown("""
""", unsafe_allow_html=True)
# ── Constants ──────────────────────────────────────────
HF_TOKEN = os.getenv("HF_TOKEN","")
JINA_KEY = os.getenv("JINA_KEY","")
PASSWORD = "132733yy"
HF_API = "https://router.huggingface.co/v1/chat/completions"
MODELS = {
"ceo": "deepseek-ai/DeepSeek-V3-0324:novita",
"researcher": "moonshotai/Kimi-K2-Instruct:novita",
"coder": "Qwen/Qwen3-Coder-480B-A35B-Instruct:novita",
"tutor": "meta-llama/Llama-3.1-8B-Instruct:cerebras",
"reasoner": "deepseek-ai/DeepSeek-R1:novita",
}
INFO = {
"ceo": {"icon":"✦", "label":"Orchestrator","model":"DeepSeek V3"},
"researcher": {"icon":"🌐", "label":"Researcher", "model":"Kimi K2"},
"coder": {"icon":"⌨️", "label":"Coder", "model":"Qwen3-Coder"},
"tutor": {"icon":"📖", "label":"Tutor", "model":"Llama 3.1"},
"reasoner": {"icon":"🧠", "label":"Reasoner", "model":"DeepSeek R1"},
}
PROMPTS = {
"ceo":"""You are an intelligent orchestrator. Read the user message and decide which agents to use.
Reply ONLY with this JSON (no extra text):
{"plan":"one sentence","agents":[{"name":"agent","task":"specific task"}]}
Rules: simple question→tutor only | code→coder | URL→researcher | math/logic→reasoner | max 3 agents.
Always respond in the same language as the user.""",
"researcher":"""You are a precise research analyst. Respond in the user's language.
Analyze URLs and content deeply. Cite sources. Say "I'm not sure" when needed.
Format: key points → analysis → your assessment. Never fabricate.""",
"coder":"""You are an expert software engineer. Respond in the user's language.
Write clean, documented, working code. Always include error handling and usage examples.
Warn about bugs immediately. Ask for clarification if ambiguous.""",
"tutor":"""You are an exceptional educator. Respond in the user's language (Arabic→Arabic, English→English).
Start with a real-world analogy, then explain from simple to complex.
Never fabricate. End with a check question.""",
"reasoner":"""You are a deep reasoning AI. Respond in the user's language.
Solve complex problems step by step. Show your reasoning clearly.
Compare solutions when multiple exist. Accuracy over speed.""",
}
MODES = [("💬","Chat"),("🔍","Search"),("⌨️","Code"),("🎨","Image"),("🎙️","Voice"),("🎬","Video")]
# ── Login ──────────────────────────────────────────────
def check_password():
if st.session_state.get("auth"): return True
st.markdown('',unsafe_allow_html=True)
c1,c2,c3=st.columns([1,1.2,1])
with c2:
pwd=st.text_input("",placeholder="Password...",type="password",key="pw",label_visibility="collapsed")
if st.button("Continue →",use_container_width=True):
if hmac.compare_digest(pwd,PASSWORD):
st.session_state["auth"]=True; st.rerun()
else: st.error("Incorrect password")
return False
if not check_password(): st.stop()
# ── Helpers ────────────────────────────────────────────
def fetch_url(url):
try:
h={"Accept":"text/plain"}
if JINA_KEY: h["Authorization"]=f"Bearer {JINA_KEY}"
r=requests.get(f"https://r.jina.ai/{url}",headers=h,timeout=15)
return r.text[:4000] if r.status_code==200 else f"[HTTP {r.status_code}]"
except Exception as e: return f"[Error:{e}]"
def extract_urls(t): return re.findall(r'https?://[^\s]+',t)
def call_llm(messages,model_key,max_tokens=1200):
if not HF_TOKEN: return "⚠️ HF_TOKEN missing in Space secrets."
try:
r=requests.post(HF_API,
headers={"Authorization":f"Bearer {HF_TOKEN}","Content-Type":"application/json"},
json={"model":MODELS[model_key],"messages":messages,"max_tokens":max_tokens,"temperature":0.7,"stream":False},
timeout=90)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
except requests.exceptions.HTTPError:
c=r.status_code
if c==503: return "⏳ Model loading, retry in a moment."
if c==429: return "🚦 Rate limit, please wait."
return f"❌ HTTP {c}: {r.text[:200]}"
except Exception as e: return f"❌ Error: {e}"
def run_agent(name,task,ctx):
msgs=[{"role":"system","content":PROMPTS[name]},{"role":"user","content":f"Task: {task}\n\nContext:\n{ctx}"}]
return name, call_llm(msgs,name)
def generate_image(prompt):
try:
r=requests.post("https://api-inference.huggingface.co/models/black-forest-labs/FLUX.1-schnell",
headers={"Authorization":f"Bearer {HF_TOKEN}"},
json={"inputs":prompt},timeout=60)
if r.status_code==200:
return "data:image/png;base64,"+base64.b64encode(r.content).decode()
except: pass
return None
def generate_voice(text):
try:
r=requests.post("https://api-inference.huggingface.co/models/microsoft/speecht5_tts",
headers={"Authorization":f"Bearer {HF_TOKEN}"},
json={"inputs":text[:400]},timeout=60)
if r.status_code==200: return r.content
except: pass
return None
# ── Pipeline ───────────────────────────────────────────
def pipeline(user_msg, history, sbox, mode):
enriched=user_msg
urls=extract_urls(user_msg)
if urls:
with sbox: st.markdown('Reading URLs with Jina...
',unsafe_allow_html=True)
for u in urls[:2]: enriched+=f"\n\n[URL:{u}]\n{fetch_url(u)}"
if mode=="image":
with sbox: st.markdown('Generating image with FLUX.1...
',unsafe_allow_html=True)
img=generate_image(user_msg); sbox.empty()
return ("IMG",img) if img else ("TXT","⚠️ Image generation failed — model may be loading. Try again in 30s."), [], {}
if mode=="voice":
with sbox: st.markdown('Generating voice...
',unsafe_allow_html=True)
audio=generate_voice(user_msg); sbox.empty()
return (("AUDIO",audio) if audio else ("TXT","⚠️ Voice generation failed.")), [], {}
if mode=="video":
sbox.empty()
return ("TXT","🎬 For video generation, use [Kling AI](https://klingai.com) — free & high quality. Open in new tab, describe your scene, download the result."), [], {}
# CEO plan
with sbox: st.markdown('Orchestrator analyzing...
',unsafe_allow_html=True)
hist_txt="\n".join([f"{m['role']}: {m['content'][:150]}" for m in history[-6:]])
ceo_msgs=[{"role":"system","content":PROMPTS["ceo"]},{"role":"user","content":f"History:\n{hist_txt}\n\nMessage:\n{enriched}"}]
raw=call_llm(ceo_msgs,"ceo",512)
try:
m=re.search(r'\{.*\}',raw,re.DOTALL)
d=json.loads(m.group()) if m else {}
agents=d.get("agents",[{"name":"tutor","task":enriched}])
except: agents=[{"name":"tutor","task":enriched}]
valid=set(MODELS)-{"ceo"}
agents=[a for a in agents if a.get("name") in valid] or [{"name":"tutor","task":enriched}]
with sbox:
for a in agents:
i=INFO[a["name"]]
st.markdown(f'{i["icon"]} {i["label"]} — {a["task"][:50]}...
',unsafe_allow_html=True)
results={}
with ThreadPoolExecutor(max_workers=len(agents)) as ex:
futs={ex.submit(run_agent,a["name"],a["task"],enriched):a["name"] for a in agents}
for f in as_completed(futs):
n,r=f.result(); results[n]=r
with sbox: st.markdown(f'{INFO[n]["icon"]} {INFO[n]["label"]} done
',unsafe_allow_html=True)
if len(results)==1:
return ("TXT",list(results.values())[0]), agents, results
with sbox: st.markdown('Synthesizing final answer...
',unsafe_allow_html=True)
synth=[{"role":"system","content":"You are an orchestrator synthesizing team results. Respond in the user's language. Merge insights into one cohesive, well-structured response."},
{"role":"user","content":f"Question:\n{user_msg}\n\nTeam:\n"+ "\n\n".join([f"=={INFO[n]['label']}==\n{r}" for n,r in results.items()])+"\n\nWrite unified final response."}]
final=call_llm(synth,"ceo",1200)
return ("TXT",final), agents, results
# ── Session ────────────────────────────────────────────
for k,v in [("msgs",[]),("mode","chat"),("upload_ctx","")]:
if k not in st.session_state: st.session_state[k]=v
# ── Sidebar ────────────────────────────────────────────
with st.sidebar:
st.markdown("### ✦ Yai"); st.divider()
st.markdown("**Active models:**")
for k,i in INFO.items(): st.markdown(f"{i['icon']} `{i['model']}`")
st.divider()
st.markdown(f"HF Token: {'🟢' if HF_TOKEN else '🔴'}")
st.markdown(f"Jina: {'🟢' if JINA_KEY else '🟡'}")
st.divider()
show_think=st.toggle("Show reasoning",True)
show_agents=st.toggle("Show agent details",False)
st.divider()
if st.button("🗑️ Clear chat",use_container_width=True):
st.session_state.msgs=[]; st.rerun()
# ── Main ───────────────────────────────────────────────
if not st.session_state.msgs:
st.markdown("""
Y
Yai
Ask anything — 5 AI models working together
💡
Explain a concept
Any topic, any level
⌨️
Write code
Any language or framework
🔍
Research a topic
Paste a URL or ask
🧠
Solve a problem
Logic, math, analysis
""", unsafe_allow_html=True)
# Mode bar
cols=st.columns(len(MODES))
for i,(icon,label) in enumerate(MODES):
with cols[i]:
active=st.session_state.mode==label.lower()
if st.button(f"{icon} {label}",key=f"m{i}",type="primary" if active else "secondary",use_container_width=True):
st.session_state.mode=label.lower(); st.rerun()
# File upload
c1,c2=st.columns([8,1])
with c2:
if st.button("📎",help="Attach file"):
st.session_state["show_up"]=not st.session_state.get("show_up",False); st.rerun()
if st.session_state.get("show_up"):
up=st.file_uploader("",type=["txt","py","js","md","csv","png","jpg","pdf"],label_visibility="collapsed")
if up:
if up.type.startswith("image/"):
b64=base64.b64encode(up.read()).decode()
st.session_state.upload_ctx=f"[Image: {up.name}]"
st.image(f"data:{up.type};base64,{b64}",width=180)
else:
try:
content=up.read().decode("utf-8",errors="ignore")
st.session_state.upload_ctx=f"[File: {up.name}]\n{content[:3000]}"
st.success(f"✓ {up.name}")
except: st.error("Could not read file")
st.markdown("---")
# Chat history
for msg in st.session_state.msgs:
if msg["role"]=="user":
with st.chat_message("user"):
st.markdown(msg["content"])
else:
with st.chat_message("assistant"):
rt=msg.get("rt","TXT")
if rt=="IMG": st.image(msg["content"],use_container_width=True)
elif rt=="AUDIO": st.audio(msg["content"],format="audio/wav")
else:
content=msg["content"]
think=re.search(r'(.*?)',content,re.DOTALL)
if think and show_think:
with st.expander("🧠 View reasoning"):
st.markdown(f"```\n{think.group(1).strip()}\n```")
clean=re.sub(r'.*?','',content,flags=re.DOTALL).strip()
st.markdown(clean)
st.markdown('⎘↺👍👎
',unsafe_allow_html=True)
if show_agents and msg.get("details") and len(msg["details"])>1:
with st.expander("View agent responses"):
for n,r in msg["details"].items():
i=INFO.get(n,{}); st.markdown(f"**{i.get('icon','')} {i.get('label',n)}** · `{i.get('model','')}`")
st.markdown(r); st.divider()
if msg.get("agents"):
tags=" · ".join([INFO[a["name"]]["model"] for a in msg["agents"] if a["name"] in INFO])
st.markdown(f'✦ {tags}
',unsafe_allow_html=True)
# Input
pmap={"chat":"Ask Yai anything...","search":"Paste a URL or ask...","code":"Describe the code you need...","image":"Describe the image...","voice":"Text to convert to voice...","video":"Describe the video scene..."}
if user_input:=st.chat_input(pmap.get(st.session_state.mode,"Ask Yai...")):
full=user_input
if st.session_state.upload_ctx:
full+="\n\n"+st.session_state.upload_ctx
st.session_state.upload_ctx=""
st.session_state["show_up"]=False
st.session_state.msgs.append({"role":"user","content":user_input})
with st.chat_message("user"): st.markdown(user_input)
with st.chat_message("assistant"):
sbox=st.container()
history=[m for m in st.session_state.msgs if m["role"]!="system"][:-1]
result,agents,details=pipeline(full,history,sbox,st.session_state.mode)
sbox.empty()
rt,rc=result
if rt=="IMG": st.image(rc,use_container_width=True)