feat: add streaming summary with segmented SSE updates

backend/agent_summary/agent/summary_agent.py

@@ -1,6 +1,7 @@
 """SummaryAgent 核心实现"""
 
 import re
+from collections.abc import AsyncIterator
 from time import time
 from urllib.parse import urlparse
 
@@ -178,6 +179,171 @@ async def summarize(self, entry_id: str, content: str) -> dict:
             "usage": usage,
         }
 
+    async def stream_summarize(self, entry_id: str, content: str) -> AsyncIterator[dict]:
+        """流式摘要：按段推送 chunk 事件，最后推送 complete。"""
+        from ..analysis.chunker import chunk_by_headings, chunk_by_paragraphs
+        from ..analysis.prompts import SUMMARY_PROMPTS
+        from ..analysis.strategies import select_strategy
+        from ..core.config import CHUNK_MAX_CHARS
+
+        state = AgentState(entry_id=entry_id, content=content)
+
+        await self.hooks.emit("before_analyze", state)
+        state = await AnalyzeStep().execute(state, self)
+        await self.hooks.emit("after_analyze", state)
+
+        next_step = await self.router.route(state)
+        state.step_history.append(f"route:{next_step}")
+
+        if next_step == "search":
+            state = await SearchStep().execute(state, self)
+
+        strategy = select_strategy(state.profile) if state.profile else "direct"
+
+        yield {
+            "type": "start",
+            "entry_id": entry_id,
+            "provider": self.provider_name,
+            "model": self.model_name,
+            "strategy": strategy,
+        }
+
+        chunk_index = 0
+        accumulated = ""
+
+        if strategy == "hierarchical":
+            chunks = chunk_by_headings(state.content, CHUNK_MAX_CHARS)
+            chunk_summaries: list[str] = []
+            for chunk in chunks:
+                prompt = SUMMARY_PROMPTS["hierarchical"].format(content=chunk)
+                summary = await self.llm.chat(prompt)
+                chunk_summaries.append(summary)
+                accumulated = "\n\n".join(chunk_summaries)
+                yield {
+                    "type": "chunk",
+                    "chunk_index": chunk_index,
+                    "delta_text": summary,
+                    "summary_text": accumulated,
+                    "phase": "chunk",
+                    "failed": False,
+                }
+                chunk_index += 1
+
+            merge_prompt = SUMMARY_PROMPTS["merge"].format(
+                chunk_summaries="\n".join(chunk_summaries)
+            )
+            merge_text = ""
+            async for token in self._stream_llm_text(merge_prompt):
+                merge_text += token
+                yield {
+                    "type": "chunk",
+                    "chunk_index": chunk_index,
+                    "delta_text": token,
+                    "summary_text": merge_text,
+                    "phase": "merge",
+                    "failed": False,
+                }
+            state.summary = merge_text
+            chunk_index += 1
+
+        elif strategy == "single_pass":
+            chunks = chunk_by_paragraphs(state.content, CHUNK_MAX_CHARS)
+            chunk_summaries = []
+            for chunk in chunks:
+                prompt = SUMMARY_PROMPTS["hierarchical"].format(content=chunk)
+                summary = await self.llm.chat(prompt)
+                chunk_summaries.append(summary)
+                accumulated = "\n\n".join(chunk_summaries)
+                yield {
+                    "type": "chunk",
+                    "chunk_index": chunk_index,
+                    "delta_text": summary,
+                    "summary_text": accumulated,
+                    "phase": "chunk",
+                    "failed": False,
+                }
+                chunk_index += 1
+
+            merge_prompt = SUMMARY_PROMPTS["merge"].format(
+                chunk_summaries="\n".join(chunk_summaries)
+            )
+            merge_text = ""
+            async for token in self._stream_llm_text(merge_prompt):
+                merge_text += token
+                yield {
+                    "type": "chunk",
+                    "chunk_index": chunk_index,
+                    "delta_text": token,
+                    "summary_text": merge_text,
+                    "phase": "merge",
+                    "failed": False,
+                }
+            state.summary = merge_text
+            chunk_index += 1
+
+        else:
+            prompt = SUMMARY_PROMPTS["direct"].format(content=state.content)
+            direct_text = ""
+            async for token in self._stream_llm_text(prompt):
+                direct_text += token
+                yield {
+                    "type": "chunk",
+                    "chunk_index": chunk_index,
+                    "delta_text": token,
+                    "summary_text": direct_text,
+                    "phase": "direct",
+                    "failed": False,
+                }
+            state.summary = direct_text
+            chunk_index += 1
+
+        state.step_history.append(f"summarize:{strategy}")
+
+        before_eval = state.summary
+        state = await EvaluateStep().execute(state, self)
+        if state.summary != before_eval and "evaluate:retry" in state.step_history:
+            yield {
+                "type": "chunk",
+                "chunk_index": chunk_index,
+                "delta_text": state.summary or "",
+                "summary_text": state.summary or "",
+                "phase": "evaluate",
+                "failed": False,
+            }
+
+        usage = {}
+        if hasattr(self.llm, "last_usage"):
+            usage = self.llm.last_usage
+        elif "usage" in state.metadata:
+            usage = state.metadata["usage"]
+
+        yield {
+            "type": "complete",
+            "result": {
+                "entry_id": entry_id,
+                "summary_text": state.summary or "",
+                "status": "success",
+                "provider": self.provider_name,
+                "model": self.model_name,
+            },
+            "usage": usage,
+        }
+
+    async def _stream_llm_text(self, prompt: str) -> AsyncIterator[str]:
+        """优先使用 provider 的 token 流式输出，否则一次性返回。"""
+        if isinstance(self.llm, LLMProviderAdapter):
+            messages = [ChatMessage(role="user", content=prompt)]
+            result = await self.llm._provider.chat(messages, stream=True)
+            if hasattr(result, "__aiter__"):
+                async for token in result:
+                    yield token
+                return
+            yield result.content
+            return
+
+        text = await self.llm.chat(prompt)
+        yield text
+
 
 def _provider_name_from_base_url(base_url: str) -> str:
     if not base_url:

backend/agent_summary/http/router.py

@@ -1,6 +1,10 @@
 """HTTP routes for summary agent requests."""
 
+import json
+import logging
+
 from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
 
 from app.schemas.agent import SummaryRequest, SummaryResult
 
@@ -11,6 +15,8 @@
     SummaryService,
 )
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(prefix="/agents/summary", tags=["agent-summary"])
 
 
@@ -29,3 +35,41 @@ async def generate_summary(request: SummaryRequest) -> SummaryResult:
         raise HTTPException(status_code=409, detail="Entry has no summary content") from exc
     except LLMClientError as exc:
         raise HTTPException(status_code=502, detail=exc.detail) from exc
+
+
+@router.post("/stream")
+async def generate_summary_stream(request: SummaryRequest) -> StreamingResponse:
+    service = get_summary_service()
+
+    try:
+        stream = service.stream_generate(request)
+        first_event = await stream.__anext__()
+
+        async def event_stream():
+            first_payload = json.dumps(first_event, ensure_ascii=False)
+            yield f"event: {first_event.get('type', 'message')}\ndata: {first_payload}\n\n"
+            async for event in stream:
+                event_type = event.get("type", "message")
+                payload = json.dumps(event, ensure_ascii=False)
+                yield f"event: {event_type}\ndata: {payload}\n\n"
+
+        return StreamingResponse(
+            event_stream(),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+            },
+        )
+
+    except ArticleNotFoundError as exc:
+        logger.warning("Streaming summary request for missing article: %s", request.entry_id)
+        raise HTTPException(status_code=404, detail="Entry not found") from exc
+    except ArticleContentUnavailableError as exc:
+        logger.warning(
+            "Streaming summary request for article without content: %s",
+            request.entry_id,
+        )
+        raise HTTPException(status_code=409, detail="Entry has no summary content") from exc
+    except LLMClientError as exc:
+        raise HTTPException(status_code=502, detail=exc.detail) from exc

backend/agent_summary/service.py

@@ -2,13 +2,14 @@
 
 import os
 import re
+from collections.abc import AsyncIterator
 from dataclasses import dataclass
-from datetime import UTC
+from datetime import UTC, datetime
 
 from app.schemas.agent import SummaryRequest, SummaryResult
 from db import get_article, get_article_content, record_usage, save_agent_result
 from llm_providers import ChatMessage, get_provider
-from llm_providers.base import LLMProvider
+from llm_providers.base import LLMProvider, LLMProviderError, ProviderNotFoundError
 
 from .agent.summary_agent import SummaryAgent
 
@@ -67,6 +68,101 @@ async def generate(self, request: SummaryRequest) -> SummaryResult:
 
         return summary
 
+    async def stream_generate(self, request: SummaryRequest) -> AsyncIterator[dict]:
+        article = get_article(request.entry_id)
+        if article is None:
+            raise ArticleNotFoundError(request.entry_id)
+
+        content = self._resolve_content(request.entry_id, article.reader_html)
+        if not content:
+            raise ArticleContentUnavailableError(request.entry_id)
+
+        async for event in self._execute_summary_stream(
+            entry_id=request.entry_id,
+            content=content,
+            request=request,
+        ):
+            yield event
+
+    async def _execute_summary_stream(
+        self,
+        entry_id: str,
+        content: str,
+        request: SummaryRequest,
+    ) -> AsyncIterator[dict]:
+        provider_name = request.provider
+        model_name = request.model
+
+        try:
+            agent = self._build_agent(request)
+            provider_name = agent.provider_name
+            model_name = agent.model_name
+
+            async for event in agent.stream_summarize(entry_id, content):
+                if event.get("type") != "complete":
+                    yield event
+                    continue
+
+                usage = event.get("usage", {})
+                result_payload = event["result"]
+                summary = SummaryResult(
+                    entry_id=result_payload["entry_id"],
+                    summary_text=result_payload["summary_text"],
+                    status=result_payload["status"],
+                    provider=result_payload["provider"],
+                    model=result_payload["model"],
+                )
+                save_agent_result(summary)
+
+                if usage:
+                    today = datetime.now(UTC).strftime("%Y-%m-%d")
+                    record_usage(
+                        day=today,
+                        provider=summary.provider,
+                        model=summary.model,
+                        agent="summary",
+                        prompt_tokens=usage.get("prompt_tokens", 0),
+                        completion_tokens=usage.get("completion_tokens", 0),
+                    )
+
+                yield {
+                    "type": "complete",
+                    "result": summary.model_dump(),
+                }
+
+        except ProviderNotFoundError:
+            result = SummaryResult(
+                entry_id=entry_id,
+                summary_text="",
+                status="failure",
+                provider=provider_name or "unknown",
+                model=model_name or "unknown",
+            )
+            save_agent_result(result)
+            yield {"type": "complete", "result": result.model_dump()}
+
+        except LLMProviderError:
+            result = SummaryResult(
+                entry_id=entry_id,
+                summary_text="",
+                status="failure",
+                provider=provider_name or "unknown",
+                model=model_name or "unknown",
+            )
+            save_agent_result(result)
+            yield {"type": "complete", "result": result.model_dump()}
+
+        except Exception:
+            result = SummaryResult(
+                entry_id=entry_id,
+                summary_text="",
+                status="failure",
+                provider=provider_name or "unknown",
+                model=model_name or "unknown",
+            )
+            save_agent_result(result)
+            yield {"type": "complete", "result": result.model_dump()}
+
     def _build_agent(self, request: SummaryRequest) -> SummaryAgent:
         if self.agent_factory is not SummaryAgent and not request.provider and not request.model:
             return _instantiate_agent(self.agent_factory)

backend/agent_summary/tests/test_agent/test_summary_agent.py

@@ -64,3 +64,13 @@ async def test_mock_llm(self):
         llm = MockLLM()
         response = await llm.chat("Test prompt")
         assert response == "Test prompt"
+
+    @pytest.mark.anyio
+    async def test_stream_summarize_emits_chunk_events(self, short_article):
+        agent = SummaryAgent(use_mock=True)
+        events = [event async for event in agent.stream_summarize("test-stream", short_article)]
+
+        assert events[0]["type"] == "start"
+        assert any(event["type"] == "chunk" for event in events)
+        assert events[-1]["type"] == "complete"
+        assert events[-1]["result"]["summary_text"]