diff --git a/AGENTS.md b/AGENTS.md index 1f85a5a..2a64fe7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1 +1,2 @@ -你编写具有可维护性和高可读性的代码，尽量不去对pipecat框架本身修改 \ No newline at end of file +你编写具有可维护性、高可读性、模块化的的代码，尽量不去对pipecat框架本身修改 +以MVP构建为目标 \ No newline at end of file diff --git a/backend/app.py b/backend/app.py index 299dc47..8926e59 100644 --- a/backend/app.py +++ b/backend/app.py @@ -23,6 +23,7 @@ from fastapi.middleware.cors import CORSMiddleware from routes import ( assistants, auth, + conversations, health, knowledge_bases, model_registry, @@ -52,6 +53,7 @@ app.add_middleware( app.include_router(health.router) app.include_router(auth.router) +app.include_router(conversations.router) app.include_router(assistants.router) app.include_router(knowledge_bases.router) app.include_router(model_registry.router) diff --git a/backend/db/models.py b/backend/db/models.py index 6b48714..0ec206e 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -9,7 +9,17 @@ from datetime import datetime -from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, String, func +from sqlalchemy import ( + JSON, + Boolean, + DateTime, + ForeignKey, + Integer, + String, + Text, + UniqueConstraint, + func, +) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column @@ -187,3 +197,83 @@ class AssistantToolBinding(Base): created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) + + +class ConversationSession(Base): + """一次完整连接对应的对话会话；媒体文件以后作为 artifact 关联进来。""" + + __tablename__ = "conversation_sessions" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + assistant_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("assistants.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + assistant_name: Mapped[str] = mapped_column(String(128), default="") + channel: Mapped[str] = mapped_column(String(24), index=True) + runtime_mode: Mapped[str] = mapped_column(String(16), default="pipeline") + status: Mapped[str] = mapped_column(String(16), index=True, default="active") + message_count: Mapped[int] = mapped_column(Integer, default=0) + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), index=True + ) + ended_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + extra: Mapped[dict] = mapped_column(JSONB, default=dict) + + +class ConversationMessage(Base): + """会话中的有序消息；当前只写 text，结构预留其他内容类型。""" + + __tablename__ = "conversation_messages" + __table_args__ = ( + UniqueConstraint( + "session_id", "sequence", name="uq_conversation_message_sequence" + ), + ) + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + session_id: Mapped[str] = mapped_column( + String(40), + ForeignKey("conversation_sessions.id", ondelete="CASCADE"), + index=True, + ) + sequence: Mapped[int] = mapped_column(Integer) + role: Mapped[str] = mapped_column(String(16)) + content_type: Mapped[str] = mapped_column(String(16), default="text") + content: Mapped[str] = mapped_column(Text, default="") + occurred_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + extra: Mapped[dict] = mapped_column(JSONB, default=dict) + + +class ConversationArtifact(Base): + """未来的音频、视频、图片等大对象索引；二进制本体不进入数据库。""" + + __tablename__ = "conversation_artifacts" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + session_id: Mapped[str] = mapped_column( + String(40), + ForeignKey("conversation_sessions.id", ondelete="CASCADE"), + index=True, + ) + message_id: Mapped[str | None] = mapped_column( + String(40), + ForeignKey("conversation_messages.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + kind: Mapped[str] = mapped_column(String(16), index=True) + storage_uri: Mapped[str] = mapped_column(String(1024)) + mime_type: Mapped[str] = mapped_column(String(128), default="") + size_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True) + duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + extra: Mapped[dict] = mapped_column(JSONB, default=dict) diff --git a/backend/migrations/versions/20260710_0003_add_conversation_history.py b/backend/migrations/versions/20260710_0003_add_conversation_history.py new file mode 100644 index 0000000..afc4ed2 --- /dev/null +++ b/backend/migrations/versions/20260710_0003_add_conversation_history.py @@ -0,0 +1,81 @@ +"""add conversation history + +Revision ID: 20260710_0003 +Revises: 20260710_0002 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "20260710_0003" +down_revision: str | Sequence[str] | None = "20260710_0002" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "conversation_sessions", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("assistant_id", sa.String(length=40), nullable=True), + sa.Column("assistant_name", sa.String(length=128), server_default="", nullable=False), + sa.Column("channel", sa.String(length=24), nullable=False), + sa.Column("runtime_mode", sa.String(length=16), server_default="pipeline", nullable=False), + sa.Column("status", sa.String(length=16), server_default="active", nullable=False), + sa.Column("message_count", sa.Integer(), server_default="0", nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("extra", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.ForeignKeyConstraint(["assistant_id"], ["assistants.id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_conversation_sessions_assistant_id", "conversation_sessions", ["assistant_id"]) + op.create_index("ix_conversation_sessions_channel", "conversation_sessions", ["channel"]) + op.create_index("ix_conversation_sessions_status", "conversation_sessions", ["status"]) + op.create_index("ix_conversation_sessions_started_at", "conversation_sessions", ["started_at"]) + + op.create_table( + "conversation_messages", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("session_id", sa.String(length=40), nullable=False), + sa.Column("sequence", sa.Integer(), nullable=False), + sa.Column("role", sa.String(length=16), nullable=False), + sa.Column("content_type", sa.String(length=16), server_default="text", nullable=False), + sa.Column("content", sa.Text(), server_default="", nullable=False), + sa.Column("occurred_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("extra", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.ForeignKeyConstraint(["session_id"], ["conversation_sessions.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("session_id", "sequence", name="uq_conversation_message_sequence"), + ) + op.create_index("ix_conversation_messages_session_id", "conversation_messages", ["session_id"]) + + op.create_table( + "conversation_artifacts", + sa.Column("id", sa.String(length=40), nullable=False), + sa.Column("session_id", sa.String(length=40), nullable=False), + sa.Column("message_id", sa.String(length=40), nullable=True), + sa.Column("kind", sa.String(length=16), nullable=False), + sa.Column("storage_uri", sa.String(length=1024), nullable=False), + sa.Column("mime_type", sa.String(length=128), server_default="", nullable=False), + sa.Column("size_bytes", sa.Integer(), nullable=True), + sa.Column("duration_ms", sa.Integer(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("extra", postgresql.JSONB(astext_type=sa.Text()), server_default=sa.text("'{}'::jsonb"), nullable=False), + sa.ForeignKeyConstraint(["message_id"], ["conversation_messages.id"], ondelete="SET NULL"), + sa.ForeignKeyConstraint(["session_id"], ["conversation_sessions.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_conversation_artifacts_session_id", "conversation_artifacts", ["session_id"]) + op.create_index("ix_conversation_artifacts_message_id", "conversation_artifacts", ["message_id"]) + op.create_index("ix_conversation_artifacts_kind", "conversation_artifacts", ["kind"]) + + +def downgrade() -> None: + op.drop_table("conversation_artifacts") + op.drop_table("conversation_messages") + op.drop_table("conversation_sessions") diff --git a/backend/routes/conversations.py b/backend/routes/conversations.py new file mode 100644 index 0000000..f387e65 --- /dev/null +++ b/backend/routes/conversations.py @@ -0,0 +1,117 @@ +"""对话历史查询 API。""" + +from db.models import ConversationMessage, ConversationSession +from db.session import get_session +from fastapi import APIRouter, Depends, HTTPException, Query +from schemas import ( + ConversationDetailOut, + ConversationListOut, + ConversationMessageOut, + ConversationOut, +) +from services.auth import require_admin +from sqlalchemy import func, or_, select +from sqlalchemy.ext.asyncio import AsyncSession + + +router = APIRouter( + prefix="/api/conversations", + tags=["conversations"], + dependencies=[Depends(require_admin)], +) + + +def _session_out(row: ConversationSession) -> ConversationOut: + return ConversationOut( + id=row.id, + assistant_id=row.assistant_id, + assistant_name=row.assistant_name, + channel=row.channel, + runtime_mode=row.runtime_mode, + status=row.status, + message_count=row.message_count, + started_at=row.started_at, + ended_at=row.ended_at, + ) + + +@router.get("", response_model=ConversationListOut) +async def list_conversations( + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), + assistant_id: str | None = None, + search: str | None = Query(None, max_length=128), + channel: str | None = None, + status: str | None = None, + sort_order: str = Query("newest", pattern="^(newest|oldest)$"), + session: AsyncSession = Depends(get_session), +): + filters = [] + if assistant_id: + filters.append(ConversationSession.assistant_id == assistant_id) + if search and search.strip(): + keyword = f"%{search.strip()}%" + filters.append( + or_( + ConversationSession.assistant_name.ilike(keyword), + ConversationSession.id.ilike(keyword), + ) + ) + if channel: + filters.append(ConversationSession.channel == channel) + if status: + filters.append(ConversationSession.status == status) + total = await session.scalar( + select(func.count()).select_from(ConversationSession).where(*filters) + ) + rows = ( + await session.execute( + select(ConversationSession) + .where(*filters) + .order_by( + ConversationSession.started_at.asc() + if sort_order == "oldest" + else ConversationSession.started_at.desc() + ) + .offset((page - 1) * page_size) + .limit(page_size) + ) + ).scalars().all() + return ConversationListOut( + items=[_session_out(row) for row in rows], + total=int(total or 0), + page=page, + page_size=page_size, + ) + + +@router.get("/{conversation_id}", response_model=ConversationDetailOut) +async def get_conversation( + conversation_id: str, + session: AsyncSession = Depends(get_session), +): + conversation = await session.get(ConversationSession, conversation_id) + if not conversation: + raise HTTPException(404, "对话记录不存在") + messages = ( + await session.execute( + select(ConversationMessage) + .where(ConversationMessage.session_id == conversation_id) + .order_by(ConversationMessage.sequence) + ) + ).scalars().all() + return ConversationDetailOut( + **_session_out(conversation).model_dump(), + messages=[ + ConversationMessageOut( + id=message.id, + sequence=message.sequence, + role=message.role, + content_type=message.content_type, + content=message.content, + occurred_at=message.occurred_at, + extra=message.extra or {}, + ) + for message in messages + ], + ) diff --git a/backend/routes/voice_webrtc.py b/backend/routes/voice_webrtc.py index bf5c834..658b8b6 100644 --- a/backend/routes/voice_webrtc.py +++ b/backend/routes/voice_webrtc.py @@ -120,7 +120,13 @@ async def _handle_offer(websocket, payload, peers): video_in_enabled=vision_enabled, ) asyncio.create_task( - run_pipeline(transport, cfg, vision_enabled=vision_enabled) + run_pipeline( + transport, + cfg, + vision_enabled=vision_enabled, + assistant_id=offer.assistant_id, + channel="webrtc", + ) ) answer = pc.get_answer() diff --git a/backend/routes/voice_ws.py b/backend/routes/voice_ws.py index 5ce5a22..eeb57ec 100644 --- a/backend/routes/voice_ws.py +++ b/backend/routes/voice_ws.py @@ -24,13 +24,16 @@ from starlette.websockets import WebSocketDisconnect router = APIRouter() -async def _resolve_start_config(raw: str) -> AssistantConfig: +async def _resolve_start_config(raw: str) -> tuple[AssistantConfig, str | None]: data = json.loads(raw) if data.get("assistant_id"): async with SessionLocal() as session: - return await resolve_runtime_config(session, data["assistant_id"]) + return ( + await resolve_runtime_config(session, data["assistant_id"]), + data["assistant_id"], + ) if data.get("inline_config"): - return AssistantConfig(**data["inline_config"]) + return AssistantConfig(**data["inline_config"]), None raise ValueError("启动参数缺少 assistant_id 或 inline_config") @@ -43,10 +46,10 @@ async def voice_stream(websocket: WebSocket): return await websocket.accept() try: - cfg = await _resolve_start_config(await websocket.receive_text()) + cfg, assistant_id = await _resolve_start_config(await websocket.receive_text()) transport = build_ws_transport(websocket) # 直接 await:管线持续读这条 WS 的音频帧,直到对端断开 - await run_pipeline(transport, cfg) + await run_pipeline(transport, cfg, assistant_id=assistant_id, channel="websocket") except WebSocketDisconnect: logger.info("WS 音频流断开") except Exception as e: diff --git a/backend/schemas.py b/backend/schemas.py index 7d7ae7b..562bef8 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -7,6 +7,7 @@ JSON 用 camelCase(modelId/interfaceType/apiUrl/apiKey),Python 内部用 snake_c from __future__ import annotations +from datetime import datetime from typing import Annotated, Any, Literal, Union from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator @@ -199,3 +200,37 @@ class ModelResourceTestResult(CamelModel): latency_ms: int | None = None message: str detail: str = "" + + +# ---------- 对话历史 ---------- +class ConversationMessageOut(CamelModel): + id: str + sequence: int + role: Literal["user", "assistant"] + content_type: str + content: str + occurred_at: datetime + extra: dict[str, Any] = Field(default_factory=dict) + + +class ConversationOut(CamelModel): + id: str + assistant_id: str | None + assistant_name: str + channel: str + runtime_mode: str + status: str + message_count: int + started_at: datetime + ended_at: datetime | None + + +class ConversationDetailOut(ConversationOut): + messages: list[ConversationMessageOut] = Field(default_factory=list) + + +class ConversationListOut(CamelModel): + items: list[ConversationOut] + total: int + page: int + page_size: int diff --git a/backend/services/conversation_history.py b/backend/services/conversation_history.py new file mode 100644 index 0000000..a2b5aa9 --- /dev/null +++ b/backend/services/conversation_history.py @@ -0,0 +1,137 @@ +"""对话历史持久化。 + +只依赖管线已经发给客户端的最终文本事件，不侵入 Pipecat。媒体历史以后写入 +conversation_artifacts，并把对象存储地址关联到会话或消息。 +""" + +import asyncio +from datetime import UTC, datetime +from uuid import uuid4 + +from db.models import ConversationMessage, ConversationSession +from db.session import SessionLocal +from loguru import logger + + +def _parse_timestamp(value: object) -> datetime: + if not isinstance(value, str) or not value: + return datetime.now(UTC) + try: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) + except ValueError: + return datetime.now(UTC) + + +class ConversationRecorder: + """按事件顺序写入一通会话；写库失败不应中断实时通话。""" + + def __init__(self, session_id: str): + self.session_id = session_id + self._sequence = 0 + self._lock = asyncio.Lock() + self._seen_events: set[str] = set() + + @classmethod + async def start( + cls, + *, + assistant_id: str | None, + assistant_name: str, + channel: str, + runtime_mode: str, + ) -> "ConversationRecorder | None": + session_id = f"conv_{uuid4().hex[:20]}" + try: + async with SessionLocal() as db: + db.add( + ConversationSession( + id=session_id, + assistant_id=assistant_id, + assistant_name=assistant_name, + channel=channel, + runtime_mode=runtime_mode, + status="active", + message_count=0, + extra={}, + ) + ) + await db.commit() + return cls(session_id) + except Exception as exc: + logger.error(f"创建对话历史会话失败，不影响本次通话: {exc}") + return None + + async def record_transport_message(self, message: object) -> None: + if not isinstance(message, dict): + return + event_type = message.get("type") + role = "" + content = "" + extra: dict = {} + timestamp = message.get("timestamp") + event_key = "" + + if event_type == "transcript": + role = str(message.get("role") or "") + content = str(message.get("content") or "").strip() + event_key = f"transcript:{role}:{timestamp}:{content}" + elif event_type == "assistant-text-end": + role = "assistant" + content = str(message.get("content") or "").strip() + turn_id = str(message.get("turn_id") or "") + event_key = f"assistant:{turn_id}" + extra = { + "turn_id": turn_id, + "interrupted": bool(message.get("interrupted", False)), + } + else: + return + + if role not in {"user", "assistant"} or not content or event_key in self._seen_events: + return + self._seen_events.add(event_key) + await self._append(role, content, timestamp, extra) + + async def _append( + self, + role: str, + content: str, + timestamp: object, + extra: dict, + ) -> None: + async with self._lock: + next_sequence = self._sequence + 1 + try: + async with SessionLocal() as db: + db.add( + ConversationMessage( + id=f"msg_{uuid4().hex[:20]}", + session_id=self.session_id, + sequence=next_sequence, + role=role, + content_type="text", + content=content, + occurred_at=_parse_timestamp(timestamp), + extra=extra, + ) + ) + conversation = await db.get(ConversationSession, self.session_id) + if conversation: + conversation.message_count = next_sequence + await db.commit() + self._sequence = next_sequence + except Exception as exc: + logger.error(f"保存对话文本失败，不影响本次通话: {exc}") + + async def finish(self, *, status: str = "completed") -> None: + try: + async with SessionLocal() as db: + conversation = await db.get(ConversationSession, self.session_id) + if conversation: + conversation.status = status + conversation.ended_at = datetime.now(UTC) + conversation.message_count = self._sequence + await db.commit() + except Exception as exc: + logger.error(f"结束对话历史会话失败: {exc}") diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 923ede3..d95315f 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -17,6 +17,7 @@ from models import AssistantConfig from openai import AsyncOpenAI from PIL import Image from services.brains import build_brain +from services.conversation_history import ConversationRecorder from services.pipecat.service_factory import ( create_realtime_service, create_stt, @@ -298,6 +299,20 @@ class RealtimeTextInputProcessor(FrameProcessor): ) +class ConversationHistoryProcessor(FrameProcessor): + """从最终客户端事件旁路保存历史，不改变 Pipecat 的上下文与帧语义。""" + + def __init__(self, recorder: ConversationRecorder | None): + super().__init__() + self._recorder = recorder + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + await self.push_frame(frame, direction) + if self._recorder and isinstance(frame, OutputTransportMessageUrgentFrame): + await self._recorder.record_transport_message(frame.message) + + class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): """聚合 LLM 回复进上下文,同时继续把回复帧交给下游 TTS。""" @@ -363,6 +378,8 @@ async def run_pipeline( cfg: AssistantConfig, *, vision_enabled: bool = False, + assistant_id: str | None = None, + channel: str = "webrtc", ) -> None: """在给定 transport 上构建并运行管线,直到连接结束。 @@ -388,7 +405,12 @@ async def run_pipeline( if cfg.runtimeMode == "realtime": if vision_enabled: logger.warning("Realtime 模式暂未接入视频帧工具,本次仅启用语音通话") - await run_realtime_pipeline(transport, cfg) + await run_realtime_pipeline( + transport, + cfg, + assistant_id=assistant_id, + channel=channel, + ) return stt = create_stt(cfg) @@ -677,6 +699,12 @@ async def run_pipeline( reason = str(call_end_state["reason"] or "completed") await queue_call_end(reason) + recorder = await ConversationRecorder.start( + assistant_id=assistant_id, + assistant_name=cfg.name, + channel=channel, + runtime_mode=cfg.runtimeMode, + ) pipeline = Pipeline( [ transport.input(), @@ -691,6 +719,7 @@ async def run_pipeline( assistant_aggregator, tts, EndCallAfterSpeech(), + ConversationHistoryProcessor(recorder), transport.output(), ] ) @@ -950,21 +979,42 @@ async def run_pipeline( await worker.queue_frame(EndFrame()) runner = WorkerRunner(handle_sigint=False) - await runner.add_workers(worker) - await runner.run() + run_status = "completed" + try: + await runner.add_workers(worker) + await runner.run() + except Exception: + run_status = "failed" + raise + finally: + if recorder: + await recorder.finish(status=run_status) logger.info("管线已结束") -async def run_realtime_pipeline(transport, cfg: AssistantConfig) -> None: +async def run_realtime_pipeline( + transport, + cfg: AssistantConfig, + *, + assistant_id: str | None = None, + channel: str = "webrtc", +) -> None: """Run a speech-to-speech model that owns ASR, reasoning, and synthesis.""" realtime = create_realtime_service(cfg) text_input = RealtimeTextInputProcessor() + recorder = await ConversationRecorder.start( + assistant_id=assistant_id, + assistant_name=cfg.name, + channel=channel, + runtime_mode=cfg.runtimeMode, + ) pipeline = Pipeline( [ transport.input(), text_input, realtime, + ConversationHistoryProcessor(recorder), transport.output(), ] ) @@ -1017,6 +1067,14 @@ async def run_realtime_pipeline(transport, cfg: AssistantConfig) -> None: await worker.queue_frame(EndFrame()) runner = WorkerRunner(handle_sigint=False) - await runner.add_workers(worker) - await runner.run() + run_status = "completed" + try: + await runner.add_workers(worker) + await runner.run() + except Exception: + run_status = "failed" + raise + finally: + if recorder: + await recorder.finish(status=run_status) logger.info("Realtime 管线已结束") diff --git a/frontend/src/components/pages/HistoryPage.tsx b/frontend/src/components/pages/HistoryPage.tsx index 03c410e..54497e2 100644 --- a/frontend/src/components/pages/HistoryPage.tsx +++ b/frontend/src/components/pages/HistoryPage.tsx @@ -1,10 +1,442 @@ -import { PlaceholderPage } from "./PlaceholderPage"; +"use client"; + +import { + ChevronDown, + ChevronUp, + Eye, + Loader2, + MessageSquareText, +} from "lucide-react"; +import { useCallback, useEffect, useMemo, useState } from "react"; + +import { Badge } from "@/components/ui/badge"; +import { Button } from "@/components/ui/button"; +import { DataList, type DataListColumn } from "@/components/ui/data-list"; +import { + Dialog, + DialogClose, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { FilterPills } from "@/components/ui/filter-pills"; +import { ListToolbar } from "@/components/ui/list-toolbar"; +import { PageHeader } from "@/components/ui/page-header"; +import { SearchInput } from "@/components/ui/search-input"; +import { + conversationsApi, + type Conversation, + type ConversationDetail, +} from "@/lib/api"; +import { cn } from "@/lib/utils"; + +const PAGE_SIZE = 20; +const channelFilters = ["全部", "WebRTC", "WebSocket"] as const; +type ChannelFilter = (typeof channelFilters)[number]; +type SortOrder = "newest" | "oldest"; + +function formatDate(value?: string | null): string { + if (!value) return "—"; + const date = new Date(value); + if (Number.isNaN(date.getTime())) return "—"; + return date.toLocaleString("zh-CN", { + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + }); +} + +function durationLabel(row: Conversation): string { + if (!row.endedAt) return "进行中"; + const seconds = Math.max( + 0, + Math.round( + (new Date(row.endedAt).getTime() - new Date(row.startedAt).getTime()) / + 1000, + ), + ); + const minutes = Math.floor(seconds / 60); + return minutes ? `${minutes} 分 ${seconds % 60} 秒` : `${seconds} 秒`; +} + +function channelLabel(channel: string): string { + if (channel === "webrtc") return "WebRTC"; + if (channel === "websocket") return "WebSocket"; + return channel; +} + +function channelValue(filter: ChannelFilter): string | undefined { + if (filter === "WebRTC") return "webrtc"; + if (filter === "WebSocket") return "websocket"; + return undefined; +} + +function statusLabel(status: string): string { + if (status === "active") return "进行中"; + if (status === "failed") return "失败"; + return "已结束"; +} export function HistoryPage() { + const [rows, setRows] = useState([]); + const [total, setTotal] = useState(0); + const [currentPage, setCurrentPage] = useState(1); + const [search, setSearch] = useState(""); + const [filter, setFilter] = useState("全部"); + const [sortOrder, setSortOrder] = useState("newest"); + const [loading, setLoading] = useState(true); + const [loadError, setLoadError] = useState(""); + + const [dialogOpen, setDialogOpen] = useState(false); + const [detail, setDetail] = useState(null); + const [detailLoading, setDetailLoading] = useState(false); + const [detailError, setDetailError] = useState(""); + + const load = useCallback(async () => { + setLoading(true); + setLoadError(""); + try { + const result = await conversationsApi.list({ + page: currentPage, + pageSize: PAGE_SIZE, + search: search.trim() || undefined, + channel: channelValue(filter), + sortOrder, + }); + setRows(result.items); + setTotal(result.total); + } catch (error) { + setLoadError(error instanceof Error ? error.message : "加载历史记录失败"); + } finally { + setLoading(false); + } + }, [currentPage, filter, search, sortOrder]); + + useEffect(() => { + // Initial and query-driven external API synchronization. + // eslint-disable-next-line react-hooks/set-state-in-effect + void load(); + }, [load]); + + const openDetail = useCallback(async (conversation: Conversation) => { + setDialogOpen(true); + setDetail(null); + setDetailError(""); + setDetailLoading(true); + try { + setDetail(await conversationsApi.get(conversation.id)); + } catch (error) { + setDetailError( + error instanceof Error ? error.message : "对话内容加载失败", + ); + } finally { + setDetailLoading(false); + } + }, []); + + function changeSearch(value: string) { + setSearch(value); + setCurrentPage(1); + } + + function changeFilter(value: ChannelFilter) { + setFilter(value); + setCurrentPage(1); + } + + function toggleSortOrder() { + setSortOrder((value) => (value === "newest" ? "oldest" : "newest")); + setCurrentPage(1); + } + + const columns = useMemo[]>( + () => [ + { + key: "assistant", + header: "助手名称", + width: "md:w-[360px]", + cell: (row) => ( + <> +
+ {row.assistantName || "调试会话"} +
+
+ {row.id} +
+ + ), + }, + { + key: "channel", + header: "接入通道", + width: "md:w-[150px]", + cell: (row) => ( + + {channelLabel(row.channel)} + + ), + }, + { + key: "status", + header: "状态", + width: "md:w-[120px]", + cell: (row) => ( + + {statusLabel(row.status)} + + ), + }, + { + key: "messages", + header: "消息数", + width: "md:w-[100px]", + cellClassName: "tabular-nums text-muted-foreground", + cell: (row) => `${row.messageCount} 条`, + }, + { + key: "duration", + header: "对话时长", + width: "md:w-[130px]", + cellClassName: "tabular-nums text-muted-foreground", + cell: durationLabel, + }, + { + key: "startedAt", + width: "md:w-[190px]", + header: ( + + ), + cellClassName: + "whitespace-nowrap tabular-nums text-muted-foreground", + cell: (row) => formatDate(row.startedAt), + }, + { + key: "actions", + header: "操作", + align: "right", + cellClassName: "flex justify-end", + cell: (row) => ( + + ), + }, + ], + [openDetail, sortOrder], + ); + + const totalPages = Math.max(1, Math.ceil(total / PAGE_SIZE)); + const safeCurrentPage = Math.min(currentPage, totalPages); + const pageStart = (safeCurrentPage - 1) * PAGE_SIZE; + const pageEnd = Math.min(pageStart + rows.length, total); + return ( - +
+ + +
+ + } + search={ + + } + /> + + row.id} + loading={loading} + error={loadError || null} + onRetry={() => void load()} + onRowClick={(row) => void openDetail(row)} + empty={{ + title: + total === 0 && !search && filter === "全部" + ? "暂无对话记录" + : "未找到匹配的对话记录", + description: + total === 0 && !search && filter === "全部" + ? "完成一次助手通话后，文本历史会显示在这里。" + : "请调整关键词或筛选条件后再试。", + }} + pagination={{ + page: safeCurrentPage, + totalPages, + onPageChange: setCurrentPage, + summary: + total === 0 + ? "没有数据" + : `显示 ${pageStart + 1}-${pageEnd} / 共 ${total} 次会话`, + }} + /> +
+ + + + + + + {detail?.assistantName || "对话详情"} + + + {detail + ? `${formatDate(detail.startedAt)} · ${channelLabel(detail.channel)} · ${detail.messageCount} 条消息` + : "查看本次会话中最终确认的文本消息。"} + + + +
+ {detailLoading && ( +
+ + 正在加载对话内容 +
+ )} + + {!detailLoading && detailError && ( +
+ {detailError} +
+ )} + + {detail && ( +
+
+
+ 会话信息 +
+
+ + + + +
+
+ +
+
+ 对话内容 +
+
+ {detail.messages.map((message) => ( +
+
+
+ {message.role === "user" ? "用户" : "助手"} ·{" "} + {formatDate(message.occurredAt)} +
+
+ {message.content} +
+ {message.extra.interrupted && ( +
+ 回复在生成中被打断 +
+ )} +
+
+ ))} + {detail.messages.length === 0 && ( +
+ 本次会话没有产生文本消息 +
+ )} +
+
+
+ )} +
+ + + + + + + + +
+ ); +} + +function Metadata({ label, value }: { label: string; value: string }) { + return ( +
+
{label}
+
{value}
+
); } diff --git a/frontend/src/hooks/use-voice-preview.ts b/frontend/src/hooks/use-voice-preview.ts index 7f46bde..a93753a 100644 --- a/frontend/src/hooks/use-voice-preview.ts +++ b/frontend/src/hooks/use-voice-preview.ts @@ -230,6 +230,7 @@ export function useVoicePreview( const dataChannelRef = useRef(null); const localStreamRef = useRef(null); const startingRef = useRef(false); + const negotiationInProgressRef = useRef(false); const messageSeqRef = useRef(0); const networkStatsRef = useRef(null); const videoBitrateRef = useRef(INITIAL_VIDEO_BITRATE); @@ -341,6 +342,7 @@ export function useVoicePreview( localStreamRef.current = null; if (audioRef.current) audioRef.current.srcObject = null; startingRef.current = false; + negotiationInProgressRef.current = false; onNodeActiveRef.current?.(null); }, []); @@ -436,10 +438,18 @@ export function useVoicePreview( try { const msg = JSON.parse(event.data); if (msg.type === "answer") { - await pcRef.current?.setRemoteDescription({ - type: "answer", - sdp: msg.payload.sdp, - }); + const currentPc = pcRef.current; + if (!currentPc) return; + try { + await currentPc.setRemoteDescription({ + type: "answer", + sdp: msg.payload.sdp, + }); + } finally { + if (pcRef.current === currentPc) { + negotiationInProgressRef.current = false; + } + } } else if (msg.type === "ice-candidate" && msg.payload?.candidate) { // 后端当前不主动 trickle,留兼容 try { @@ -478,6 +488,42 @@ export function useVoicePreview( const pc = new RTCPeerConnection({ iceServers }); pcRef.current = pc; + const sendOffer = async () => { + if ( + pcRef.current !== pc || + ws.readyState !== WebSocket.OPEN || + pc.signalingState !== "stable" || + negotiationInProgressRef.current + ) { + return; + } + + negotiationInProgressRef.current = true; + try { + const offer = await pc.createOffer(); + await pc.setLocalDescription(offer); + const localDescription = pc.localDescription; + if (!localDescription?.sdp) { + throw new Error("浏览器无法创建 WebRTC offer。"); + } + ws.send( + JSON.stringify({ + type: "offer", + payload: { + pc_id: pcId, + sdp: localDescription.sdp, + type: localDescription.type, + assistant_id: assistantId, + vision_enabled: Boolean(options.visionEnabled), + }, + }), + ); + } catch (offerError) { + negotiationInProgressRef.current = false; + throw offerError; + } + }; + pc.onicecandidate = (e) => { if (ws.readyState !== WebSocket.OPEN) return; ws.send( @@ -503,10 +549,15 @@ export function useVoicePreview( channel.onopen = () => { channel.send(JSON.stringify({ type: "client-ready" })); }; - channel.onmessage = (event) => { + channel.onmessage = async (event) => { try { const msg = JSON.parse(event.data); if ( + msg?.type === "signalling" && + msg?.message?.type === "renegotiate" + ) { + await sendOffer(); + } else if ( msg?.type === "assistant-text-start" && typeof msg.turn_id === "string" ) { @@ -631,25 +682,8 @@ export function useVoicePreview( } } - // 4) 生成 offer 并发给后端(assistant_id 在 payload 顶层) - const offer = await pc.createOffer(); - await pc.setLocalDescription(offer); - const localDescription = pc.localDescription; - if (!localDescription?.sdp) { - throw new Error("浏览器无法创建 WebRTC offer。"); - } - ws.send( - JSON.stringify({ - type: "offer", - payload: { - pc_id: pcId, - sdp: localDescription.sdp, - type: localDescription.type, - assistant_id: assistantId, - vision_enabled: Boolean(options.visionEnabled), - }, - }), - ); + // 4) 生成 offer 并发给后端；同一方法也响应 SmallWebRTC 的重新协商请求。 + await sendOffer(); } catch (connectionError) { fail(errorMessage(connectionError, "无法连接语音服务。")); } finally { diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index a791f0a..0c5d6e4 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -194,6 +194,63 @@ export const assistantsApi = { request<{ ok: boolean }>(`/api/assistants/${id}`, { method: "DELETE" }), }; +// ---------- 对话历史 ---------- +export type Conversation = { + id: string; + assistantId: string | null; + assistantName: string; + channel: string; + runtimeMode: string; + status: string; + messageCount: number; + startedAt: string; + endedAt: string | null; +}; + +export type ConversationMessage = { + id: string; + sequence: number; + role: "user" | "assistant"; + contentType: string; + content: string; + occurredAt: string; + extra: { interrupted?: boolean; turn_id?: string }; +}; + +export type ConversationDetail = Conversation & { + messages: ConversationMessage[]; +}; + +export type ConversationList = { + items: Conversation[]; + total: number; + page: number; + pageSize: number; +}; + +export const conversationsApi = { + list: (params: { + page?: number; + pageSize?: number; + search?: string; + channel?: string; + status?: string; + sortOrder?: "newest" | "oldest"; + } = {}) => { + const query = new URLSearchParams({ + page: String(params.page ?? 1), + page_size: String(params.pageSize ?? 20), + sort_order: params.sortOrder ?? "newest", + }); + if (params.search) query.set("search", params.search); + if (params.channel) query.set("channel", params.channel); + if (params.status) query.set("status", params.status); + return request(`/api/conversations?${query}`); + }, + get: (id: string) => + request(`/api/conversations/${id}`), +}; + // ---------- 工具 ---------- export type ToolStatus = "active" | "archived" | "draft"; export type ToolParameter = {