feat:优化流式输出,增加临时多轮对话记录
This commit is contained in:
@@ -0,0 +1,30 @@
|
||||
"""会话创建与历史查询接口"""
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
|
||||
from app.common.response import ApiResponse, success_response
|
||||
from app.core.exceptions import BizException
|
||||
from app.schemas.conversation import ConversationCreated, ConversationDetail
|
||||
|
||||
router = APIRouter(prefix="/conversations", tags=["Conversation"])
|
||||
|
||||
|
||||
@router.post("")
|
||||
async def create_conversation(
|
||||
request: Request,
|
||||
) -> ApiResponse[ConversationCreated]:
|
||||
store = request.app.state.conversation_store
|
||||
return success_response(ConversationCreated(conversation_id=await store.create()))
|
||||
|
||||
|
||||
@router.get("/{conversation_id}")
|
||||
async def get_conversation(
|
||||
conversation_id: str, request: Request
|
||||
) -> ApiResponse[ConversationDetail]:
|
||||
store = request.app.state.conversation_store
|
||||
if not await store.exists(conversation_id):
|
||||
raise BizException(404, "会话不存在或已过期")
|
||||
turns = await store.load_turns(conversation_id, None)
|
||||
return success_response(
|
||||
ConversationDetail(conversation_id=conversation_id, turns=turns)
|
||||
)
|
||||
@@ -0,0 +1,27 @@
|
||||
"""会话及对话轮次,用于多轮上下文和后续持久化"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class Turn(BaseModel):
|
||||
"""一次任务对应一轮对话,同一个 thread_id 多次写入时覆盖同一轮"""
|
||||
|
||||
conversation_id: str
|
||||
thread_id: str
|
||||
message_id: str
|
||||
prompt: str # 用户原始输入
|
||||
outcome: Literal["completed", "waiting_for_input", "failed", "cancelled"]
|
||||
summary: str # 助手侧摘要,只描述后端生成了什么,不代表前端已执行
|
||||
created_at: datetime
|
||||
|
||||
|
||||
class ConversationCreated(BaseModel):
|
||||
conversation_id: str
|
||||
|
||||
|
||||
class ConversationDetail(BaseModel):
|
||||
conversation_id: str
|
||||
turns: list[Turn]
|
||||
@@ -0,0 +1,71 @@
|
||||
"""会话存储:创建时即占位,用于校验 ID 由后端签发;当前为进程内存实现"""
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Protocol
|
||||
from uuid import uuid4
|
||||
|
||||
from app.config import settings
|
||||
from app.schemas.conversation import Turn
|
||||
|
||||
|
||||
class ConversationStore(Protocol):
|
||||
"""后续接入 MongoDB 时实现同一组方法,业务代码不需要改动"""
|
||||
|
||||
async def create(self) -> str: ...
|
||||
|
||||
async def exists(self, conversation_id: str) -> bool: ...
|
||||
|
||||
async def load_turns(
|
||||
self, conversation_id: str, limit: int | None
|
||||
) -> list[Turn]: ...
|
||||
|
||||
async def save_turn(self, turn: Turn) -> None: ...
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(UTC)
|
||||
|
||||
|
||||
class InMemoryConversationStore:
|
||||
def __init__(self, empty_ttl: int | None = None):
|
||||
ttl = empty_ttl if empty_ttl is not None else settings.conversation_empty_ttl
|
||||
self._empty_ttl = timedelta(seconds=ttl)
|
||||
# 会话 ID -> 创建时间;轮次按 thread_id 去重并保持首次写入顺序
|
||||
self._created: dict[str, datetime] = {}
|
||||
self._turns: dict[str, dict[str, Turn]] = {}
|
||||
|
||||
def _purge_expired(self) -> None:
|
||||
# 只清理没有任何轮次且超过保留时长的空会话,有对话的会话不受影响
|
||||
deadline = _now() - self._empty_ttl
|
||||
for conversation_id in [
|
||||
key
|
||||
for key, created in self._created.items()
|
||||
if created < deadline and not self._turns.get(key)
|
||||
]:
|
||||
del self._created[conversation_id]
|
||||
self._turns.pop(conversation_id, None)
|
||||
|
||||
async def create(self) -> str:
|
||||
self._purge_expired()
|
||||
conversation_id = str(uuid4())
|
||||
self._created[conversation_id] = _now()
|
||||
return conversation_id
|
||||
|
||||
async def exists(self, conversation_id: str) -> bool:
|
||||
self._purge_expired()
|
||||
return conversation_id in self._created
|
||||
|
||||
async def load_turns(self, conversation_id: str, limit: int | None) -> list[Turn]:
|
||||
turns = list(self._turns.get(conversation_id, {}).values())
|
||||
turns.sort(key=lambda turn: turn.created_at)
|
||||
if limit is None:
|
||||
return turns
|
||||
# 只取最近 limit 轮,limit 为 0 时不带历史
|
||||
return turns[-limit:] if limit else []
|
||||
|
||||
async def save_turn(self, turn: Turn) -> None:
|
||||
# 同一任务再次写入(澄清恢复后)保留首次创建时间,只更新内容
|
||||
existing = self._turns.setdefault(turn.conversation_id, {}).get(turn.thread_id)
|
||||
if existing is not None:
|
||||
turn = turn.model_copy(update={"created_at": existing.created_at})
|
||||
self._turns[turn.conversation_id][turn.thread_id] = turn
|
||||
Reference in New Issue
Block a user