流式翻译

backend/agent_translation/agent.py

@@ -2,6 +2,8 @@
 
 import asyncio
 import re
+from collections.abc import AsyncIterator
+from dataclasses import dataclass
 
 from llm_providers import (
     ChatMessage,
@@ -55,6 +57,12 @@ def chunk_by_headings(text: str, max_chars: int = 4000) -> list[str]:
     return chunks if chunks else [text]
 
 
+@dataclass(slots=True)
+class BilingualSegment:
+    kind: str
+    source_text: str
+
+
 class TranslationAgent:
     """
     Agent for translating article content to a target language.
@@ -164,58 +172,20 @@ async def translate_bilingual(
         Returns:
             Dictionary with translated_text (bilingual format) and usage
         """
-        # 按段落分段
-        paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]
-
-        # 合并短段落，避免太碎片化
-        merged_paragraphs = []
-        current = ""
-        for para in paragraphs:
-            if len(current) + len(para) < 500:
-                current = current + "\n\n" + para if current else para
-            else:
-                if current:
-                    merged_paragraphs.append(current)
-                current = para
-        if current:
-            merged_paragraphs.append(current)
-
         total_prompt_tokens = 0
         total_completion_tokens = 0
         bilingual_parts = []
         failed_chunks = 0
 
-        for para in merged_paragraphs:
-            # 检查是否是标题
-            if para.startswith('#'):
-                # 标题直接翻译，不保留原文；失败则退化为原标题
-                try:
-                    result = await self.translate_chunk(para, target_lang, temperature)
-                    bilingual_parts.append(result["text"])
-                    total_prompt_tokens += result["prompt_tokens"]
-                    total_completion_tokens += result["completion_tokens"]
-                except LLMProviderError:
-                    failed_chunks += 1
-                    bilingual_parts.append(para)
-            else:
-                # 普通段落：先原文，再翻译。单段失败不影响整篇，保留原文占位。
-                original = f'<div class="bilingual-original">{para}</div>'
-                try:
-                    result = await self.translate_chunk(para, target_lang, temperature)
-                    translated = result["text"]
-                    total_prompt_tokens += result["prompt_tokens"]
-                    total_completion_tokens += result["completion_tokens"]
-                except LLMProviderError:
-                    failed_chunks += 1
-                    translated = para
-                translation = f'<div class="bilingual-translation">{translated}</div>'
-                bilingual_parts.append(original)
-                bilingual_parts.append("")
-                bilingual_parts.append(translation)
-                bilingual_parts.append("")
+        async for event in self.stream_translate_bilingual(content, target_lang, temperature):
+            bilingual_parts.append(event["html"])
+            total_prompt_tokens += event["prompt_tokens"]
+            total_completion_tokens += event["completion_tokens"]
+            failed_chunks += 1 if event["failed"] else 0
 
         # 全部片段都失败时，视为整体失败，交由 service 标记 failure
-        if merged_paragraphs and failed_chunks == len(merged_paragraphs):
+        segments = self.prepare_bilingual_segments(content)
+        if segments and failed_chunks == len(segments):
             raise LLMProviderError("All translation chunks failed")
 
         # 合并结果
@@ -231,6 +201,83 @@ async def translate_bilingual(
             },
         }
 
+    def prepare_bilingual_segments(self, content: str) -> list[BilingualSegment]:
+        paragraphs = [p.strip() for p in content.split('\n\n') if p.strip()]
+
+        merged_paragraphs = []
+        current = ""
+        for para in paragraphs:
+            if len(current) + len(para) < 500:
+                current = current + "\n\n" + para if current else para
+            else:
+                if current:
+                    merged_paragraphs.append(current)
+                current = para
+        if current:
+            merged_paragraphs.append(current)
+
+        segments = []
+        for para in merged_paragraphs:
+            kind = "heading" if para.startswith('#') else "paragraph"
+            segments.append(BilingualSegment(kind=kind, source_text=para))
+        return segments
+
+    async def stream_translate_bilingual(
+        self,
+        content: str,
+        target_lang: str,
+        temperature: float = 0.3,
+    ) -> AsyncIterator[dict]:
+        segments = self.prepare_bilingual_segments(content)
+
+        for index, segment in enumerate(segments):
+            if segment.kind == "heading":
+                try:
+                    result = await self.translate_chunk(
+                        segment.source_text,
+                        target_lang,
+                        temperature,
+                    )
+                    translated = result["text"]
+                    failed = False
+                    prompt_tokens = result["prompt_tokens"]
+                    completion_tokens = result["completion_tokens"]
+                except LLMProviderError:
+                    translated = segment.source_text
+                    failed = True
+                    prompt_tokens = 0
+                    completion_tokens = 0
+                yield {
+                    "chunk_index": index,
+                    "html": translated,
+                    "failed": failed,
+                    "prompt_tokens": prompt_tokens,
+                    "completion_tokens": completion_tokens,
+                }
+                continue
+
+            original = f'<div class="bilingual-original">{segment.source_text}</div>'
+            try:
+                result = await self.translate_chunk(segment.source_text, target_lang, temperature)
+                translated = result["text"]
+                failed = False
+                prompt_tokens = result["prompt_tokens"]
+                completion_tokens = result["completion_tokens"]
+            except LLMProviderError:
+                translated = segment.source_text
+                failed = True
+                prompt_tokens = 0
+                completion_tokens = 0
+
+            translation = f'<div class="bilingual-translation">{translated}</div>'
+            yield {
+                "chunk_index": index,
+                "html": f"{original}\n\n{translation}",
+                "failed": failed,
+                "prompt_tokens": prompt_tokens,
+                "completion_tokens": completion_tokens,
+            }
+
     async def translate(
         self,
         content: str,

backend/agent_translation/http/router.py

@@ -8,9 +8,11 @@
 When translation fails, a 200 response with FAILURE status is returned instead.
 """
 
+import json
 import logging
 
 from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
 
 from app.schemas.agent import TranslationRequest, TranslationResult
 
@@ -92,3 +94,40 @@ async def translate_article(request: TranslationRequest) -> TranslationResult:
     except ArticleContentUnavailableError as exc:
         logger.warning(f"Translation request for article without content: {request.entry_id}")
         raise HTTPException(status_code=409, detail="Entry has no content to translate") from exc
+
+
+@router.post("/stream")
+async def translate_article_stream(request: TranslationRequest) -> StreamingResponse:
+    service = get_translation_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(f"Streaming translation request for missing article: {request.entry_id}")
+        raise HTTPException(status_code=404, detail="Entry not found") from exc
+
+    except ArticleContentUnavailableError as exc:
+        logger.warning(
+            "Streaming translation request for article without content: %s",
+            request.entry_id,
+        )
+        raise HTTPException(status_code=409, detail="Entry has no content to translate") from exc

backend/agent_translation/service.py

@@ -8,6 +8,7 @@
 """
 
 import re
+from collections.abc import AsyncIterator
 from dataclasses import dataclass
 from datetime import UTC
 
@@ -90,6 +91,24 @@ async def generate(self, request: TranslationRequest) -> TranslationResult:
 
         return result
 
+    async def stream_generate(self, request: TranslationRequest) -> 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_translation_stream(
+            entry_id=request.entry_id,
+            target_lang=request.target_lang,
+            content=content,
+            provider_name=request.provider,
+            model_name=request.model,
+        ):
+            yield event
+
     async def _execute_translation(
         self,
         entry_id: str,
@@ -185,6 +204,138 @@ async def _execute_translation(
                 model=model_name or "unknown",
             )
 
+    async def _execute_translation_stream(
+        self,
+        entry_id: str,
+        target_lang: str,
+        content: str,
+        provider_name: str | None,
+        model_name: str | None,
+    ) -> AsyncIterator[dict]:
+        try:
+            provider = get_provider(name=provider_name)
+            agent = TranslationAgent(provider=provider)
+            segments = agent.prepare_bilingual_segments(content)
+
+            accumulated_html = ""
+            total_prompt_tokens = 0
+            total_completion_tokens = 0
+            failed_chunks = 0
+            total_chunks = 0
+            resolved_segments: dict[int, str] = {}
+
+            yield {
+                "type": "start",
+                "entry_id": entry_id,
+                "target_lang": target_lang,
+                "provider": provider.name,
+                "model": provider.model,
+            }
+
+            async for chunk in agent.stream_translate_bilingual(
+                content=content,
+                target_lang=target_lang,
+                temperature=0.3,
+            ):
+                total_chunks += 1
+                failed_chunks += 1 if chunk["failed"] else 0
+                total_prompt_tokens += chunk["prompt_tokens"]
+                total_completion_tokens += chunk["completion_tokens"]
+                accumulated_html = (
+                    f"{accumulated_html}\n\n{chunk['html']}" if accumulated_html else chunk["html"]
+                )
+                resolved_segments[chunk["chunk_index"]] = chunk["html"]
+                yield {
+                    "type": "chunk",
+                    "chunk_index": chunk["chunk_index"],
+                    "delta_html": chunk["html"],
+                    "translation_html": self._build_stream_preview_html(
+                        segments,
+                        resolved_segments,
+                    ),
+                    "failed": chunk["failed"],
+                }
+
+            if total_chunks and failed_chunks == total_chunks:
+                raise LLMProviderError("All translation chunks failed")
+
+            from datetime import datetime
+
+            today = datetime.now(UTC).strftime("%Y-%m-%d")
+            record_usage(
+                day=today,
+                provider=provider.name,
+                model=provider.model,
+                agent="translation",
+                prompt_tokens=total_prompt_tokens,
+                completion_tokens=total_completion_tokens,
+            )
+
+            result = TranslationResult(
+                entry_id=entry_id,
+                target_lang=target_lang,
+                translation_html=accumulated_html,
+                status="success",
+                provider=provider.name,
+                model=provider.model,
+            )
+            save_agent_result(result)
+            yield {
+                "type": "complete",
+                "result": result.model_dump(),
+            }
+
+        except ProviderNotFoundError:
+            result = TranslationResult(
+                entry_id=entry_id,
+                target_lang=target_lang,
+                translation_html="",
+                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 = TranslationResult(
+                entry_id=entry_id,
+                target_lang=target_lang,
+                translation_html="",
+                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 = TranslationResult(
+                entry_id=entry_id,
+                target_lang=target_lang,
+                translation_html="",
+                status="failure",
+                provider=provider_name or "unknown",
+                model=model_name or "unknown",
+            )
+            save_agent_result(result)
+            yield {"type": "complete", "result": result.model_dump()}
+
+    @staticmethod
+    def _build_stream_preview_html(segments, resolved_segments: dict[int, str]) -> str:
+        parts: list[str] = []
+        for index, segment in enumerate(segments):
+            if index in resolved_segments:
+                parts.append(resolved_segments[index])
+                continue
+
+            if segment.kind == "heading":
+                parts.append(segment.source_text)
+            else:
+                parts.append(f'<div class="bilingual-original">{segment.source_text}</div>')
+
+        return "\n\n".join(parts)
+
     def _resolve_content(self, entry_id: str, reader_html: str) -> str:
         """
         Resolve article content, preferring cleaned content over raw HTML.