core_logic_hybrid.py

# ./core_logic_hybrid.py -> Token-safe

"""

Hybrid: Local LLM with HF UI

"Master Stroke" for sharing app while keeping compute costs at zero; with UI on Hugging Face, the app "calls home" - the local PC - for answers.

We expose local Ollama, via the secret "LOCAL_LLM_URL" as "The Tunnel", a secure bridge between the Hugging Face-hosted UI and the local LLM. By default, Ollama only listens to localhost, so we tell it to accept external traffic from the tunnel:
. The UI sends user messages to the Tunnel, which forwards them to the local Ollama instance
. Ollama processes the request and sends the response back through the Tunnel to the UI."
"""

import os
from openai import OpenAI
from tools import web_search, parse_file

# Hybrid bridge - Sanitized URL to prevent double slashes
tunnel_url = os.getenv("LOCAL_LLM_URL", "").rstrip("/")

client = OpenAI(
    base_url=f"{tunnel_url}/v1",
    api_key="ollama"
)

model = "gemma4:latest"

SYSTEM_PROMPT = """
You are the 'Silicon Architect' — a full-stack, master-stroke creative genius in AI Engineering and Technical Architecture. 
Your goal is to provide production-grade, highly optimized solutions for web and mobile AI applications.

Expertise: Python (latest production version), Agentic Loops, FastAPI, and Scalable Architecture. 
Provide production-ready code and rigorous technical research with appropriate comments. Analyze files when provided. Be concise.

CORE DIRECTIVES:
1. ARCHITECTURAL RIGOR: Always consider scalability, async patterns, and state management.
2. AGENTIC EXPERTISE: You understand recurrent-depth simulations, tool-calling, and autonomous loops.
3. CODE QUALITY: Write clean, PEP 8 compliant, and secure Python/JS code.
4. INNOVATION: Suggest the latest libraries and frameworks (FastAPI, LangGraph, Pydantic AI; but not limited to these).
5. RESEARCH: If the user asks about new tech, use your Web Search capability to provide factual, up-to-date documentation.

PERSONALITY:
1. FRANK/POLITE: Disagree with the user, if needed; never resort to sycophancy, and suggest better alternatives.
2. HUMBLE: Apologize when mistaken.
3. FIRST PRINCIPLES: Base your responses and reasoning in Richard Feynman’s first principles thinking. Break down complex problems into fundamental truths and reason up from there.

When a user provides files, analyze the code structure and logic before proposing changes.
"""

def chat_function(message, history):
    user_text = message.get("text", "")
    files = message.get("files", [])
    
    # 1. Process Files with character limits
    context_from_files = ""
    for f in files:
        path = f["path"] if isinstance(f, dict) else f
        file_content = parse_file(path)
        context_from_files += file_content
    
    # TRUNCATE FILE CONTEXT: Max ~3000 tokens (approx 12,000 chars)
    if len(context_from_files) > 12000:
        context_from_files = context_from_files[:12000] + "\n...[File Content Truncated]..."

    # 2. Research Trigger
    if any(keyword in user_text.lower() for keyword in ["search", "docs", "latest"]):
        research_context = web_search(user_text)
        prompt = f"RESEARCH:\n{research_context}\n\nFILES:\n{context_from_files}\n\nUSER: {user_text}"
    else:
        prompt = f"FILES:\n{context_from_files}\n\nUSER: {user_text}"

    # 3. Build Messages with History Slicing
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    
    # Keep last 3 turns for context stability
    for turn in history[-3:]:
        messages.append({"role": turn["role"], "content": turn["content"]})
        
    messages.append({"role": "user", "content": prompt})

    try:
        completion = client.chat.completions.create(
            model=model,
            messages=messages,
            stream=True,
            temperature=0.2, # Zero for architectural precision; incremented for creative architecture
            max_tokens=1024 
        )

        response_text = ""
        for chunk in completion:
            # Check for valid delta content to avoid metadata crashes
            if chunk.choices and hasattr(chunk.choices[0].delta, 'content'):
                token = chunk.choices[0].delta.content
                if token:
                    response_text += token
                    yield response_text
    except Exception as e:
        yield f"Silicon Error: {str(e)}"