Spaces:
Sleeping
Sleeping
File size: 17,433 Bytes
e549219 e84fe91 e549219 e84fe91 e549219 59ded42 e549219 59ded42 e549219 e84fe91 e549219 e84fe91 e549219 95ceb25 e549219 59ded42 e549219 e84fe91 e549219 | 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 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 | """Service to interact with LLM using AIpipe OpenAI-compatible API."""
from typing import Dict, Any
import os
import json
import httpx
from pydantic import BaseModel, Field
from dotenv import load_dotenv
load_dotenv()
class LLMResponse(BaseModel):
"""Model for LLM response"""
readme: str = Field(..., alias="README.md")
html: str = Field(..., alias="index.html")
css: str | None = Field(None, alias="style.css")
javascript: str | None = Field(None, alias="script.js")
class AIpipeClient:
def __init__(self, model: str):
self.model = model if model is not None else "qwen/qwen3-coder"
self.api_key = os.environ.get("OPENAI_API_KEY")
self.base_url = os.environ.get("OPENAI_BASE_URL", "https://aipipe.org/openrouter/v1/chat/completions")
if not self.api_key:
raise RuntimeError("OPENAI_API_KEY (AIpipe token) not set")
async def generate_app_files(self, brief: str, checks: list[str], attachments_meta: list[dict[str, Any]]) -> Dict[str, str]:
"""
Returns a mapping of filepath -> content for the generated app.
Uses a strict JSON schema to ensure consistent output format.
"""
system = (
"You are an expert web developer. Generate professional, secure, and robust web applications following these principles:"
"\n1. DATA HANDLING BEST PRACTICES:"
"\n - Handle provided attachments based on their MIME types"
"\n - Support common formats (CSV, JSON, text, base64, data URLs)"
"\n - For URLs and data URIs: Never modify or truncate - use complete URLs as provided"
"\n - Process attachments based on their MIME types and content"
"\n - Support both direct data and URL-based resources"
"\n - Implement proper streaming for large data"
"\n - Handle encoding/decoding appropriately (base64, UTF-8, etc.)"
"\n - Never hardcode data - always read from provided sources"
"\n - Implement proper error handling for data loading/parsing"
"\n\nAttachment handling contract:"
"\n - Inputs may be data: URLs or http(s) URLs. Never modify or truncate URLs."
"\n - For data: URLs, parse header: data:<mime>[;base64],<payload>. Extract mime (default text/plain)."
"\n - If ;base64, decode to bytes then decode text via UTF-8 (TextDecoder) for text types; if not base64, URL-decode payload."
"\n - For http(s) URLs, use the exact URL including its query string. Do not reconstruct it."
"\n - CSV: trim BOM, normalize newlines, detect delimiter among [',',';','\t'] with a reasonable heuristic;"
"\n infer header row by content (treat as header only if non-numeric labels), handle quoted fields."
"\n - Text: decode UTF-8, escape before inserting into DOM, show a safe preview."
"\n - Images: assign img.src directly to the URL; for data URLs no CORS needed; for http(s) preserve query string."
"\n - Never log entire data URLs or base64 payloads; at most log lengths or short previews."
"\n - Prefer ?url= parameter if present to select the source; otherwise fall back to the first attachment."
"\n\n2. WEB DEVELOPMENT STANDARDS:"
"\n - Use semantic HTML5 elements appropriately"
"\n - Write responsive CSS using modern practices"
"\n - Make a simple but good looking frontend with HTML/CSS/JS"
"\n - Follow JavaScript best practices and security guidelines"
"\n - Support all modern browsers and devices"
"\n3. USER EXPERIENCE:"
"\n - Show loading states during async operations"
"\n - Provide clear feedback for success/failure"
"\n - Implement proper error messages and validation"
"\n - Ensure accessibility (ARIA labels, semantic markup)"
"\n4. CODE QUALITY:"
"\n - Write clean, maintainable, and documented code"
"\n - Use proper async/await patterns for data operations"
"\n - Implement comprehensive error handling"
"\n - Follow security best practices for data processing"
"\n5. TESTING & VALIDATION:"
"\n - Ensure all specified test cases pass"
"\n - Validate input data thoroughly"
"\n - Handle edge cases and errors gracefully"
"\n - Meet all requirements specified in the brief"
"\nReturn ONLY a JSON object matching this schema:"
"{"
' "README.md": "# Project Name\\n\\nProject description\\n\\n## Setup\\n...\\n\\n## Usage\\n...\\n\\n## Code Explanation\\n...\\n\\n## License\\nMIT",'
' "index.html": "<!DOCTYPE html>\\n<html lang=\\"en\\">...</html>",'
' "style.css": "/* CSS styles */",'
' "script.js": "// JavaScript code"'
"}"
"\n- Uses semantic HTML elements correctly"
"\n- Follows modern web development best practices"
"\n- Use API calls with fetch()/async/await appropriately if needed"
"\n- Treat each task(brief) as separate and independent"
)
user = {
"brief": brief,
"requirements": [
"Generate a complete, production-ready web application with a decent frontend",
"Process and validate all provided input data",
"Handle resources and attachments securely",
"Implement proper error handling and user feedback",
"Meet all specified test cases and checks",
"Follow web security and accessibility standards",
"Properly handle data from attachments and URLs, modifying the code according to their content",
"If the attachment URL is base64 encoded, decode and use the data appropriately",
"If the attachment is a data URL, parse and use the embedded data correctly",
],
"context": {
"data_handling": {
"formats": {
"csv": "Trim BOM, normalize newlines, detect delimiter among [',',';','\\t'] with a heuristic; infer header row by content instead of assuming; handle quoted fields; document column usage in comments",
"images": "Load completely, handle data URLs, support URL parameters",
"json": "Parse properly, handle nested structures",
"text": "Handle encodings, support streaming for large content"
},
"sources": {
"attachments": "Use complete URLs, respect MIME types",
"parameters": "Handle URL parameters, support dynamic loading",
"apis": "Implement proper fetch with error handling"
}
},
"attachments_spec": {
"item_shape": {"name": "string", "url": "string"},
"url_types": ["data:<mime>[;base64],<payload>", "http(s)://... with query string"],
"rules": [
"Do not modify or truncate URLs",
"For data: URLs, parse header and decode appropriately",
"For http(s), preserve full query string",
"Do not log or echo full base64 payloads",
]
},
"expected_helpers": [
"parseDataUrl(url): { mime, isBase64, text?, bytes? }",
"decodeBase64ToText(b64): string",
"parseCsv(text): { headers?: string[], rows: string[][] }"
]
},
"standards": {
"html": "Use semantic HTML5 elements and proper document structure",
"css": "Write responsive, maintainable CSS with modern features",
"javascript": "Follow ES6+ best practices, proper async handling",
"accessibility": "Implement ARIA labels and semantic markup",
"security": "Validate input, sanitize data, handle errors properly"
},
"checks": checks,
"attachments": attachments_meta,
}
url = f"{self.base_url}/chat/completions"
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": json.dumps(user, indent=2)},
],
"temperature": 0.1,
}
print("AIpipe URL:", url)
#print("AIpipe payload:", json.dumps(payload, indent=2))
print("Starting AIpipe request...")
timeouts = httpx.Timeout(
timeout=300.0,
connect=60.0,
read=300.0,
write=120.0
)
async with httpx.AsyncClient(timeout=timeouts) as client:
print("Sending request to AIpipe...")
resp = await client.post(url, headers=headers, json=payload)
print("Got response with status:", resp.status_code)
resp.raise_for_status()
print("Response is OK, getting headers...")
# print("AIpipe response headers:", dict(resp.headers))
print("Parsing JSON response...")
data = resp.json()
print("Got JSON data, extracting content...")
content = data["choices"][0]["message"]["content"]
#print("AIpipe raw response:", repr(content))
print("AIpipe processing complete")
# Clean and parse the response
content = content.strip()
# Remove any markdown code fences
content = content.removeprefix("```json").removeprefix("```").removesuffix("```").strip()
try:
# Parse as JSON first
parsed = json.loads(content)
# Add MIT license to parsed content
if "LICENSE" not in parsed:
parsed["LICENSE"] = """MIT License
Copyright (c) 2025
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE."""
# Validate against our schema
response = LLMResponse.model_validate(parsed)
# Convert to dictionary of files
files = {
"README.md": response.readme,
"index.html": response.html,
}
if response.javascript:
files["script.js"] = response.javascript
if response.css:
files["style.css"] = response.css
return files
except Exception as e:
print(f"Failed to parse LLM response: {e}")
# Fallback to a minimal app that satisfies basic structure
return {
"README.md": "# Generated App\n\nA minimal static web application.\n\n## Usage\n\nOpen index.html in a browser.\n\n## License\n\nMIT",
"LICENSE": "MIT License\n\nPermission is hereby granted...",
"index.html": """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>Generated App</title>
</head>
<body>
<h1 id="app-status">Generated App</h1>
<div id="content"></div>
<script>
document.getElementById('app-status').textContent = 'Ready';
</script>
</body>
</html>""".strip()
}
async def generate_revision_files(
self,
brief: str,
checks: list[str],
attachments_meta: list[dict[str, Any]],
current_files: Dict[str, str],
) -> Dict[str, str]:
"""Request minimal changes from the LLM. Returns mapping of only changed files."""
system = (
"You are an expert web developer performing a focused revision on an existing small static web app.\n"
"- Modify ONLY what is necessary to satisfy the new brief.\n"
"- Update README.md too if necessary to reflect changes.\n"
"- Preserve files not mentioned in your JSON output.\n"
"- Return a JSON object mapping file paths to their FULL updated contents for changed files only.\n"
"- Never truncate or alter provided URLs/data URIs; use them exactly as given.\n"
"- Keep the code simple, robust, and readable.\n"
"\nAttachment handling contract (still applies during revisions):\n"
"- Inputs may be data: URLs or http(s) URLs. Never modify or truncate URLs.\n"
"- For data: URLs, parse header: data:<mime>[;base64],<payload>. Extract mime (default text/plain).\n"
"- If ;base64, decode to bytes then decode text via UTF-8 for text types; if not base64, URL-decode payload.\n"
"- For http(s) URLs, use the exact URL including its query string. Do not reconstruct it.\n"
"- CSV: trim BOM, normalize newlines, detect delimiter among [',',';','\\t'] with a reasonable heuristic; infer header row by content; handle quoted fields.\n"
"- Text: decode UTF-8, escape before inserting into DOM.\n"
"- Images: assign img.src directly to the URL; preserve query string.\n"
"- Never log entire data URLs or base64 payloads."
)
user = {
"brief": brief,
"checks": checks,
"attachments": attachments_meta,
"current_files": {k: current_files.get(k, "") for k in [
"README.md", "index.html", "style.css", "script.js"
]},
"requirements": [
"Return only changed files in JSON (e.g., { 'README.md': '...', 'script.js': '...' })",
"Do not include unchanged files in your answer",
"Ensure output remains a working app after changes",
],
"notes": [
"If nothing needs to change, return an empty JSON object {}",
],
}
url = f"{self.base_url}/chat/completions"
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": json.dumps(user, indent=2)},
],
"temperature": 0.2,
}
timeouts = httpx.Timeout(timeout=180.0, connect=30.0, read=180.0, write=60.0)
async with httpx.AsyncClient(timeout=timeouts) as client:
resp = await client.post(url, headers=headers, json=payload)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"].strip()
content = content.removeprefix("```json").removeprefix("```").removesuffix("```").strip()
try:
parsed = json.loads(content)
if not isinstance(parsed, dict):
raise ValueError("Expected a JSON object of changed files")
# Basic sanity filter: allow only known files unless path looks safe
allowed = {"README.md", "index.html", "style.css", "script.js"}
safe_files = {k: v for k, v in parsed.items() if k in allowed or (".." not in k and "\\" not in k)}
return safe_files
except Exception as e:
print(f"Failed to parse LLM revision response: {e}")
return {}
|