Spaces:
Sleeping
Sleeping
File size: 4,744 Bytes
ae0a268 | 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 | from __future__ import annotations
import json
import os
import time
from pathlib import Path
from typing import Any, Dict
from fastapi import FastAPI, Query
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from agent_core import iter_problem_steps
ROOT_DIR = Path(__file__).resolve().parent
STATIC_DIR = ROOT_DIR / "static"
OUTPUT_DIR = ROOT_DIR / "outputs"
AGNES_MODEL = "agnes-2.0-flash"
AGNES_CHAT_COMPLETIONS_URL = "https://apihub.agnes-ai.com/v1/chat/completions"
app = FastAPI(title="General Algorithm Problem Solving Agent")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
def load_env_file(path: Path) -> None:
if not path.exists():
return
for raw_line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
line = raw_line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = value
load_env_file(ROOT_DIR / ".env")
def sse_event(payload: Dict[str, Any]) -> str:
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
@app.get("/")
def index() -> FileResponse:
return FileResponse(STATIC_DIR / "index.html")
@app.get("/api/status")
def status() -> Dict[str, Any]:
api_key = os.getenv("AGNES_API_KEY", "")
return {
"enabled": bool(api_key),
"provider": "Agnes AI",
"model": AGNES_MODEL,
"base_url_configured": True,
"workflow": [
"Memory",
"Planner",
"Retriever",
"Executor",
"Script Runner",
"Evaluator",
"Loop Controller",
"Reflector",
"Artifact Writer",
],
}
@app.get("/api/run")
def run_agent(
question: str = Query(..., min_length=1),
) -> StreamingResponse:
def stream():
selected_key = os.getenv("AGNES_API_KEY", "")
yield sse_event(
{
"type": "status",
"title": "Agent 启动",
"content": "已接收问题，开始初始化 Session Memory 与任务状态。",
"time": time.strftime("%H:%M:%S"),
}
)
try:
final_state = None
for index, (item, state) in enumerate(
iter_problem_steps(
question,
api_key=selected_key,
base_url=AGNES_CHAT_COMPLETIONS_URL,
model=AGNES_MODEL,
),
start=1,
):
final_state = state
yield sse_event(
{
"type": "step",
"index": index,
"module": item.get("module", ""),
"title": item.get("title", ""),
"content": item.get("content", ""),
"time": time.strftime("%H:%M:%S"),
}
)
time.sleep(0.05)
if final_state is None:
raise RuntimeError("Agent 未产生任何输出。")
result = {
"final_answer": final_state.final_answer,
"pdf_path": final_state.pdf_path,
"tex_path": final_state.tex_path,
}
yield sse_event(
{
"type": "final",
"title": "最终答案",
"content": result["final_answer"],
"pdf_url": "/api/report/pdf",
"tex_url": "/api/report/tex",
"time": time.strftime("%H:%M:%S"),
}
)
yield sse_event({"type": "done"})
except Exception as exc:
yield sse_event(
{
"type": "error",
"title": "运行失败",
"content": str(exc),
"time": time.strftime("%H:%M:%S"),
}
)
yield sse_event({"type": "done"})
return StreamingResponse(stream(), media_type="text/event-stream")
@app.get("/api/report/pdf")
def report_pdf() -> FileResponse:
path = OUTPUT_DIR / "solution_report.pdf"
return FileResponse(path, filename="solution_report.pdf", media_type="application/pdf")
@app.get("/api/report/tex")
def report_tex() -> FileResponse:
path = OUTPUT_DIR / "solution_report.tex"
return FileResponse(path, filename="solution_report.tex", media_type="text/plain")
|