""" 为 ``web`` 目录下的静态前端提供 API：会话列表、多模态发送（与 ``chatbot_agent`` 一致）。 从仓库 ``Agent`` 目录、且 ``PYTHONPATH`` 包含 ``src`` 时启动:: uvicorn agent.web.server:app --reload --host 127.0.0.1 --port 8765 """ from __future__ import annotations import base64 import mimetypes import os import shutil import sys import tempfile import uuid from pathlib import Path from typing import Any from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse _SRC = Path(__file__).resolve().parent.parent.parent if str(_SRC) not in sys.path: sys.path.insert(0, str(_SRC)) from agent import chatbot_agent as ca # noqa: E402 from web import ui_session as ui # noqa: E402 from agent.utils.image_utils import IMAGE_FILE_EXTENSIONS # noqa: E402 from agent.utils.agent_utils import DOC_EXT_MAP # noqa: E402 _WEB_DIR = Path(__file__).resolve().parent _IMAGE_DIR = _WEB_DIR / "image" _AUDIO_EXT = frozenset({".wav", ".mp3", ".m4a", ".flac", ".ogg", ".webm"}) _FILE_EXT = frozenset(DOC_EXT_MAP.keys()) def _data_url_for_file(fp: str) -> str | None: """将本地文件读入为 data URL，供前端直接展示（不单独提供文件下载路由）。""" if not fp or not os.path.isfile(fp): return None mime, _ = mimetypes.guess_type(fp) if not mime: ext = os.path.splitext(fp)[1].lower() if ext in IMAGE_FILE_EXTENSIONS: mime = {".png": "image/png", ".gif": "image/gif", ".webp": "image/webp", ".bmp": "image/bmp"}.get( ext, "image/jpeg" ) elif ext in _AUDIO_EXT: mime = { ".wav": "audio/wav", ".mp3": "audio/mpeg", ".m4a": "audio/mp4", ".flac": "audio/flac", ".ogg": "audio/ogg", }.get(ext, "application/octet-stream") else: mime = "application/octet-stream" try: with open(fp, "rb") as f: raw = f.read() b64 = base64.standard_b64encode(raw).decode("ascii") except OSError: return None return f"data:{mime};base64,{b64}" def _serialize_content(content: Any, *, meta: dict[str, Any] | None = None, role: str | None = None) -> Any: """将一条消息中的 content 转为 JSON 友好结构；文件以内联 data URL 输出。""" meta = meta or {} if isinstance(content, str): return content if not isinstance(content, list): return str(content) parts: list[dict[str, Any]] = [] for piece in content: if isinstance(piece, str) and piece.strip(): parts.append({"type": "text", "text": piece.strip()}) continue if not isinstance(piece, dict): continue if piece.get("type") == "text": parts.append({"type": "text", "text": piece.get("text") or ""}) continue fp = piece.get("path") or (piece.get("file") or {}).get("path") if piece.get("type") == "image_url": fp = fp or (piece.get("image_url") or {}).get("url") if piece.get("type") == "audio": fp = fp or piece.get("audio_path") if not fp or not os.path.isfile(fp): continue data_url = _data_url_for_file(fp) if not data_url: continue ext = os.path.splitext(fp)[1].lower() if ext in IMAGE_FILE_EXTENSIONS: parts.append({"type": "image", "url": data_url, "path": fp, "name": os.path.basename(fp)}) elif ext in _AUDIO_EXT: audio_meta = meta.get("audio") if isinstance(meta.get("audio"), dict) else {} parts.append({ "type": "audio", "url": data_url, "path": fp, "name": os.path.basename(fp), "autoplay": bool(meta.get("autoplay") or audio_meta.get("autoplay")) if role == "assistant" else False, "duration_seconds": audio_meta.get("duration_seconds"), }) else: parts.append({"type": "file", "url": data_url, "path": fp, "name": os.path.basename(fp)}) return parts def _serialize_messages(hist: list[dict[str, Any]] | None) -> list[dict[str, Any]]: hist = hist or [] # 找到最后一条 assistant 消息的索引，用于控制自动播放 last_assistant_idx = -1 for i in range(len(hist) - 1, -1, -1): if hist[i].get("role") == "assistant": last_assistant_idx = i break out = [] for idx, m in enumerate(hist): role = m.get("role") content = m.get("content") meta = dict(m.get("meta") or {}) meta.setdefault("output_type", "text" if role != "assistant" else meta.get("output_type") or "text") if role == "assistant": meta.setdefault("autoplay", bool((meta.get("audio") or {}).get("autoplay") or meta.get("autoplay"))) serialized_content = _serialize_content(content, meta=meta, role=role) # ---------- 音频部件处理 ---------- if role == "assistant" and meta.get("audio"): audio_meta = meta.get("audio") if isinstance(meta.get("audio"), dict) else {} # 检查 content 中是否已经包含了音频类型的部件（例如从 manifest 加载时生成的） already_has_audio = False if isinstance(serialized_content, list): for part in serialized_content: if isinstance(part, dict) and part.get("type") == "audio": already_has_audio = True # 修正历史消息的 autoplay 为 False（仅最新一条可允许） if idx != last_assistant_idx: part["autoplay"] = False break # 只有 content 中还没有音频部件时，才追加额外的 audio_part if not already_has_audio: is_latest = (idx == last_assistant_idx) autoplay_final = is_latest and (meta.get("autoplay") or audio_meta.get("autoplay")) audio_part = { "type": "audio", "url": _data_url_for_file(audio_meta.get("path")), "name": "语音回复", "autoplay": autoplay_final, "duration_seconds": audio_meta.get("duration_seconds"), } if audio_part["url"]: if isinstance(serialized_content, list): serialized_content.append(audio_part) elif serialized_content: serialized_content = [{"type": "text", "text": serialized_content}, audio_part] else: serialized_content = [audio_part] out.append({"role": role, "content": serialized_content, "meta": meta}) return out def _bootstrap() -> tuple[str, list[str], list[dict[str, Any]]]: ids = ui.list_session_ids_ordered() if ids: sid = ids[0] hist = ui.load_ui_messages_for_session(sid) else: sid = ui.new_timestamp_session_id() hist = [] ids = [sid] return sid, ids, hist app = FastAPI(title="minor web chat", version="1.0.0") app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.get("/api/bootstrap") def api_bootstrap() -> dict[str, Any]: sid, ids, hist = _bootstrap() return {"sessionId": sid, "sessions": ids, "messages": _serialize_messages(hist)} @app.post("/api/sessions/new") def api_new_session() -> dict[str, Any]: sid = ui.new_timestamp_session_id() ids = ui.list_session_ids_ordered() ids = [sid] + [x for x in ids if x != sid] return {"sessionId": sid, "sessions": ids, "messages": []} @app.get("/api/sessions/{session_id}/messages") def api_session_messages(session_id: str) -> dict[str, Any]: if not session_id: raise HTTPException(status_code=400, detail="缺少 session_id") hist = ui.load_ui_messages_for_session(session_id) ids = ui.list_session_ids_ordered() if session_id not in ids: ids = [session_id, *ids] return {"sessionId": session_id, "sessions": ids, "messages": _serialize_messages(hist)} @app.post("/api/sessions/clear") def api_clear_session(session_id: str = Form(...)) -> dict[str, Any]: if not session_id: raise HTTPException(status_code=400, detail="缺少 session_id") messages, _ = ui.clear_ui_and_manifest(session_id) ids = ui.list_session_ids_ordered() if session_id not in ids: ids = [session_id, *ids] return {"sessionId": session_id, "sessions": ids, "messages": _serialize_messages(messages)} @app.post("/api/sessions/delete") def api_delete_session(session_id: str = Form(...)) -> dict[str, Any]: if not session_id: raise HTTPException(status_code=400, detail="缺少 session_id") ui.delete_session(session_id) sid, ids, hist = _bootstrap() return {"sessionId": sid, "sessions": ids, "messages": _serialize_messages(hist)} def _materialize_uploads_to_temp( images: list[UploadFile] | None, files: list[UploadFile] | None, audio: UploadFile | None, text: str, ) -> tuple[dict[str, Any], str]: """将上传文件落盘到临时目录，返回 ``(query_dict, tmp_root)``；调用方须在 finally 中删除 ``tmp_root``。""" tmp_root = tempfile.mkdtemp(prefix="web_mm_") file_paths: list[str] = [] for uf in images or []: if uf is None or not uf.filename: continue ext = Path(uf.filename).suffix or "" dest = os.path.join(tmp_root, f"{uuid.uuid4().hex}{ext}") with open(dest, "wb") as f: shutil.copyfileobj(uf.file, f) file_paths.append(dest) for uf in files or []: if uf is None or not uf.filename: continue ext = Path(uf.filename).suffix or "" dest = os.path.join(tmp_root, f"{uuid.uuid4().hex}{ext}") with open(dest, "wb") as f: shutil.copyfileobj(uf.file, f) file_paths.append(dest) if audio is not None and audio.filename: ext = Path(audio.filename).suffix or ".bin" dest = os.path.join(tmp_root, f"{uuid.uuid4().hex}{ext}") with open(dest, "wb") as f: shutil.copyfileobj(audio.file, f) file_paths.append(dest) query = {"text": (text or "").strip(), "files": file_paths} return query, tmp_root def _chat_start_core(session_id: str, query: dict[str, Any]) -> dict[str, Any]: """追加 user、复制附件到会话目录并写 manifest（不调用模型）。""" hist = ui.load_ui_messages_for_session(session_id) before = len(hist or []) new_hist, cleared = ui.add_message(hist or [], query) if cleared is None: raise HTTPException(status_code=400, detail="请输入文字或上传图片/音频") if len(new_hist) == before: raise HTTPException(status_code=400, detail="请输入文字或上传图片/音频") ui.rewrite_user_files_for_last_turn(new_hist, session_id) ui.sync_manifest_from_chat_history(session_id, new_hist, "") ids = ui.list_session_ids_ordered() if session_id not in ids: ids = [session_id, *ids] return { "sessionId": session_id, "sessions": ids, "messages": _serialize_messages(new_hist), } def _chat_complete_core(session_id: str, output_type: str = "text") -> dict[str, Any]: """对 manifest 中末条 user 执行视觉解析与 ``chain_with_history.invoke``。""" hist = ui.load_ui_messages_for_session(session_id) if not hist or hist[-1].get("role") != "user": raise HTTPException( status_code=400, detail="没有待完成的用户消息，请先调用 /api/chat/start", ) new_hist = ca.execute_agent([*hist], session_id, output_type=output_type) ui.prune_memory_session_ids() ids = ui.list_session_ids_ordered() if session_id not in ids: ids = [session_id, *ids] return { "sessionId": session_id, "sessions": ids, "messages": _serialize_messages(new_hist), } @app.post("/api/chat/start") async def api_chat_start( session_id: str = Form(...), text: str = Form(""), images: list[UploadFile] | None = File(None), files: list[UploadFile] | None = File(None), audio: UploadFile | None = File(None), ) -> dict[str, Any]: """ 仅追加用户消息并持久化到 manifest（含附件落盘）；不调用模型。 前端应先调用本接口再调用 ``/api/chat/complete``。 """ if not session_id: raise HTTPException(status_code=400, detail="缺少 session_id") query, tmp_root = _materialize_uploads_to_temp(images, files, audio, text) try: return _chat_start_core(session_id, query) finally: shutil.rmtree(tmp_root, ignore_errors=True) @app.post("/api/chat/complete") async def api_chat_complete( session_id: str = Form(...), output_type: str = Form("text"), ) -> dict[str, Any]: """对当前会话最近一次 ``/api/chat/start`` 提交执行模型推理并返回完整消息列表。""" if not session_id: raise HTTPException(status_code=400, detail="缺少 session_id") return _chat_complete_core(session_id, output_type=output_type) @app.post("/api/chat") async def api_chat( session_id: str = Form(...), text: str = Form(""), images: list[UploadFile] | None = File(None), files: list[UploadFile] | None = File(None), audio: UploadFile | None = File(None), output_type: str = Form("text"), ) -> dict[str, Any]: """ 兼容旧客户端：等价于依次 ``/api/chat/start`` + ``/api/chat/complete``（仍阻塞至推理结束）。 """ if not session_id: raise HTTPException(status_code=400, detail="缺少 session_id") query, tmp_root = _materialize_uploads_to_temp(images, files, audio, text) try: _chat_start_core(session_id, query) return _chat_complete_core(session_id, output_type=output_type) finally: shutil.rmtree(tmp_root, ignore_errors=True) # 先注册 API，再挂静态文件，避免 ``/`` 抢路由 @app.get("/api/health") def health() -> dict[str, str]: return {"status": "ok"} @app.get("/favicon.ico") def favicon() -> FileResponse: for name, mime in ( ("favicon.ico", "image/x-icon"), ("favicon.png", "image/png"), ("favicon.svg", "image/svg+xml"), ): path = _IMAGE_DIR / name if path.is_file(): return FileResponse(path, media_type=mime) raise HTTPException(status_code=404, detail="web/image 目录下缺少 favicon.ico / favicon.png / favicon.svg") @app.get("/image/{asset_name}") def serve_image_asset(asset_name: str) -> FileResponse: path = (_IMAGE_DIR / asset_name).resolve() try: path.relative_to(_IMAGE_DIR.resolve()) except ValueError as exc: raise HTTPException(status_code=404, detail="图片资源不存在") from exc if not path.is_file(): raise HTTPException(status_code=404, detail=f"web/image 目录下缺少 {asset_name}") mime, _ = mimetypes.guess_type(str(path)) return FileResponse(path, media_type=mime or "application/octet-stream") @app.get("/") def index_page() -> FileResponse: return FileResponse(_WEB_DIR / "index.html") @app.get("/app.css") def serve_css() -> FileResponse: return FileResponse(_WEB_DIR / "app.css") @app.get("/app.js") def serve_js() -> FileResponse: return FileResponse(_WEB_DIR / "app.js")