feat: add streaming unified review assistant

server.py

@@ -8,6 +8,7 @@ import os
 import re
 import secrets
 import threading
+import time
 from datetime import date, datetime, timedelta, timezone
 from http import HTTPStatus
 from http.cookies import SimpleCookie
@@ -16,6 +17,7 @@ from typing import Any
 from urllib.parse import parse_qs, unquote, urlparse
 
 from alert_service import AlertService
+from assistant_agent import ReviewAssistantError, stream_review_assistant
 from api_access import required_role
 from app_config import (
     DATA_DIR,
@@ -1147,6 +1149,111 @@ class DashboardService:
         deleted = self.database.delete_trade_entry(self.current_user_id, trade_id)
         return {"deleted": deleted, **self.trade_entries()}
 
+    def assistant_messages(self) -> list[dict[str, Any]]:
+        return self.database.list_assistant_messages(self.current_user_id)
+
+    def clear_assistant_messages(self) -> int:
+        return self.database.delete_assistant_messages(self.current_user_id)
+
+    def assistant_stream(self, payload: dict[str, Any]):
+        question = validate_text(payload.get("question"), "问题", 2000, required=True)
+        trade_date = normalize_date(
+            str(payload.get("trade_date") or date.today().isoformat())
+        )
+        context = self._assistant_context(trade_date)
+        history = [
+            {"role": item["role"], "content": str(item["content"])[:4000]}
+            for item in self.assistant_messages()[-12:]
+            if item.get("role") in {"user", "assistant"}
+        ]
+        source = self.ensure_llm_access("assistant")
+        profiles = [
+            (
+                self.llm_primary_api_key,
+                self.llm_primary_base_url,
+                self.llm_primary_model,
+            )
+        ]
+        if self.llm_fallback_configured:
+            profiles.append(
+                (
+                    self.llm_fallback_api_key,
+                    self.llm_fallback_base_url,
+                    self.llm_fallback_model,
+                )
+            )
+
+        def generate():
+            started = time.perf_counter()
+            last_error: Exception | None = None
+            for api_key, base_url, model in profiles:
+                try:
+                    upstream = iter(
+                        stream_review_assistant(
+                            context, question, history, api_key, base_url, model
+                        )
+                    )
+                    first = next(upstream)
+                except (ReviewAssistantError, StopIteration) as exc:
+                    last_error = exc
+                    continue
+                answer_parts = [first]
+                yield first
+                try:
+                    for chunk in upstream:
+                        answer_parts.append(chunk)
+                        yield chunk
+                except ReviewAssistantError as exc:
+                    self.record_llm_usage("assistant", source, model, "failed")
+                    raise ValueError("智能解读连接中断，请稍后重试。") from exc
+                answer = "".join(answer_parts).strip()
+                latency_ms = round((time.perf_counter() - started) * 1000)
+                self.database.save_assistant_exchange(
+                    self.current_user_id, question, answer, trade_date
+                )
+                self.record_llm_usage("assistant", source, model, "success", latency_ms)
+                return
+            self.record_llm_usage("assistant", source, self.llm_primary_model, "failed")
+            raise ValueError("智能解读服务暂不可用，请稍后重试。") from last_error
+
+        return generate()
+
+    def _assistant_context(self, trade_date: str) -> dict[str, Any]:
+        dashboard = self.get_dashboard(trade_date)
+        actual_date = normalize_date(
+            str((dashboard.get("meta") or {}).get("trade_date") or trade_date)
+        )
+        sentiment = self.sentiment_history(actual_date, 10)
+        tracking = self.strategy_tracking.list_tracking(self.current_user_id, 5)
+        alerts = self.alert_service.list_alerts(
+            self.current_user_id, "all", date.today().isoformat()
+        )
+        trades = self.trade_journal.list_entries(
+            self.current_user_id, end_date=actual_date
+        )
+        return {
+            "data_date": actual_date,
+            "market": {
+                "overview": dashboard.get("overview") or {},
+                "top_sectors": (dashboard.get("sectors") or [])[:8],
+                "limit_performance": dashboard.get("limit_performance") or {},
+                "sentiment_history": (sentiment.get("rows") or [])[-10:],
+            },
+            "personal": {
+                "watchlist": self.database.list_watchlist(self.current_user_id)[:30],
+                "review_notes": self.database.list_notes(
+                    self.current_user_id, scope="daily"
+                )[:10],
+                "strategy_tracking": {
+                    "summary": tracking.get("summary") or {},
+                    "batches": (tracking.get("batches") or [])[:5],
+                },
+                "alerts": (alerts.get("items") or [])[:20],
+                "trade_summary": trades.get("summary") or {},
+                "trade_entries": (trades.get("items") or [])[:30],
+            },
+        }
+
     def sync_screener_data(self, trade_date: str, lookback: int = 45) -> dict[str, Any]:
         if not self.configured:
             raise ValueError("请先配置 Tushare Token。")
@@ -3436,6 +3543,9 @@ class RequestHandler(BaseHTTPRequestHandler):
             except ValueError as exc:
                 self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
             return
+        if parsed.path == "/api/assistant/messages":
+            self.send_json({"items": SERVICE.assistant_messages()})
+            return
         if parsed.path == "/api/dashboard":
             query = parse_qs(parsed.query)
             trade_date = query.get("trade_date", [date.today().isoformat()])[0]
@@ -3660,6 +3770,9 @@ class RequestHandler(BaseHTTPRequestHandler):
         if parsed.path == "/api/trades":
             self.save_trade_entry()
             return
+        if parsed.path == "/api/assistant/chat":
+            self.stream_assistant_chat()
+            return
         if parsed.path == "/api/admin/settings":
             self.save_system_settings()
             return
@@ -3729,6 +3842,10 @@ class RequestHandler(BaseHTTPRequestHandler):
             deleted = SERVICE.database.delete_user_birth_profile(SERVICE.current_user_id)
             self.send_json({"ok": True, "deleted": deleted})
             return
+        if parsed.path == "/api/assistant/messages":
+            deleted = SERVICE.clear_assistant_messages()
+            self.send_json({"ok": True, "deleted": deleted})
+            return
         if parsed.path == "/api/mentors/messages":
             query = parse_qs(parsed.query)
             try:
@@ -3889,6 +4006,36 @@ class RequestHandler(BaseHTTPRequestHandler):
         except (ValueError, json.JSONDecodeError) as exc:
             self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
 
+    def stream_assistant_chat(self) -> None:
+        try:
+            body = self.read_json_body()
+            stream = SERVICE.assistant_stream(body)
+        except (ValueError, json.JSONDecodeError) as exc:
+            self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
+            return
+        self.send_response(HTTPStatus.OK)
+        self.send_header("Content-Type", "application/x-ndjson; charset=utf-8")
+        self.send_header("Cache-Control", "no-cache, no-transform")
+        self.send_header("X-Accel-Buffering", "no")
+        self.send_header("Connection", "close")
+        self.end_headers()
+        try:
+            for chunk in stream:
+                self._write_stream_event({"type": "delta", "content": chunk})
+            self._write_stream_event({"type": "done"})
+        except (ValueError, ReviewAssistantError) as exc:
+            self._write_stream_event({"type": "error", "error": str(exc)})
+        except (BrokenPipeError, ConnectionResetError):
+            pass
+        finally:
+            self.close_connection = True
+
+    def _write_stream_event(self, payload: dict[str, Any]) -> None:
+        self.wfile.write(
+            (json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8")
+        )
+        self.wfile.flush()
+
     def session_token(self) -> str:
         cookie = SimpleCookie()
         try: