485 lines
16 KiB
Python
485 lines
16 KiB
Python
"""
|
|
Conversation-related router module.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Dict, Any, Optional
|
|
from urllib.parse import quote
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from fastapi.responses import FileResponse
|
|
|
|
# Removed unused imports.
|
|
from ..state import get_server_state, ServerState
|
|
from ..error_handlers import create_error_response
|
|
from ..artifacts import (
|
|
extract_latest_artifacts,
|
|
guess_content_type,
|
|
resolve_project_path,
|
|
)
|
|
from config.config import get_config
|
|
|
|
router = APIRouter(prefix="/v1/conversations", tags=["conversations"])
|
|
|
|
|
|
def _build_artifact_url(conversation_id: str, path: Path) -> str:
|
|
encoded_path = quote(str(path), safe="")
|
|
return f"/v1/conversations/{conversation_id}/artifacts/raw?path={encoded_path}"
|
|
|
|
|
|
def _read_text_artifact(path: Path) -> Optional[str]:
|
|
try:
|
|
return path.read_text(encoding="utf-8")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _serialize_artifact_file(
|
|
conversation_id: str,
|
|
path: Path,
|
|
include_content: bool = False,
|
|
) -> Dict[str, Any]:
|
|
payload: Dict[str, Any] = {
|
|
"path": str(path),
|
|
"content_type": guess_content_type(path),
|
|
"url": _build_artifact_url(conversation_id, path),
|
|
}
|
|
|
|
if include_content:
|
|
payload["content"] = _read_text_artifact(path)
|
|
|
|
return payload
|
|
|
|
|
|
@router.get(
|
|
"",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="List all available conversations",
|
|
)
|
|
async def list_conversations(state: ServerState = Depends(get_server_state)):
|
|
"""List all available conversations."""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
conversations = state.conversation_manager.list_conversations()
|
|
return {"conversations": conversations, "total_count": len(conversations)}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to list conversations: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Create a new conversation",
|
|
)
|
|
async def create_conversation(
|
|
state: ServerState = Depends(get_server_state),
|
|
):
|
|
"""Create a new conversation.
|
|
|
|
Args:
|
|
llm_interface: LLM interface name; if None, use the default interface
|
|
max_history_length: Maximum Context history length
|
|
"""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
llm_obj = get_config().CONTEXT_SUMMARY_INTERFACE
|
|
max_history_length = get_config().CONTEXT_MAX_HISTORY_LENGTH
|
|
conversation = state.conversation_manager.create_conversation(
|
|
llm_interface=llm_obj, max_history_length=max_history_length
|
|
)
|
|
|
|
return {
|
|
"conversation_id": conversation.uuid,
|
|
"created_at": conversation.created_at.isoformat(),
|
|
"last_accessed": conversation.last_accessed.isoformat(),
|
|
}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to create conversation: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{conversation_id}",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Get information for the specified conversation",
|
|
)
|
|
async def get_conversation(
|
|
conversation_id: str, state: ServerState = Depends(get_server_state)
|
|
):
|
|
"""Get information for the specified conversation."""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
conversation = state.conversation_manager.get_conversation(conversation_id)
|
|
if conversation is None:
|
|
return create_error_response(
|
|
message=f"Conversation {conversation_id} not found",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
# Get conversation statistics.
|
|
with conversation:
|
|
try:
|
|
history_count = conversation.context.get_total_message_count()
|
|
sketch_stats = conversation.sketch_pad.get_statistics()
|
|
# Convert the SketchPadStatistics object to a dictionary.
|
|
sketch_stats_dict = sketch_stats.model_dump()
|
|
except Exception as e:
|
|
history_count = 0
|
|
sketch_stats_dict = {}
|
|
|
|
return {
|
|
"conversation_id": conversation.uuid,
|
|
"created_at": conversation.created_at.isoformat(),
|
|
"last_accessed": conversation.last_accessed.isoformat(),
|
|
"message_count": history_count,
|
|
"sketch_stats": sketch_stats_dict,
|
|
}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to get conversation: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.delete(
|
|
"",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Delete all conversations",
|
|
)
|
|
async def delete_all_conversations(state: ServerState = Depends(get_server_state)):
|
|
"""Delete all conversations."""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
deleted_ids = state.conversation_manager.delete_all_conversations()
|
|
return {
|
|
"deleted": True,
|
|
"deleted_count": len(deleted_ids),
|
|
"conversation_ids": deleted_ids,
|
|
}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to delete all conversations: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.delete(
|
|
"/{conversation_id}",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Delete the specified conversation",
|
|
)
|
|
async def delete_conversation(
|
|
conversation_id: str, state: ServerState = Depends(get_server_state)
|
|
):
|
|
"""Delete the specified conversation."""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
success = state.conversation_manager.delete_conversation(conversation_id)
|
|
if not success:
|
|
return create_error_response(
|
|
message=f"Conversation {conversation_id} not found",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
return {"deleted": True, "conversation_id": conversation_id}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to delete conversation: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{conversation_id}/history",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Get the conversation history for the specified conversation",
|
|
)
|
|
async def get_conversation_history(
|
|
conversation_id: str,
|
|
limit: Optional[int] = None,
|
|
state: ServerState = Depends(get_server_state),
|
|
):
|
|
"""Get the conversation history for the specified conversation."""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
conversation = state.conversation_manager.get_conversation(conversation_id)
|
|
if conversation is None:
|
|
return create_error_response(
|
|
message=f"Conversation {conversation_id} not found",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
# Get conversation history.
|
|
with conversation:
|
|
try:
|
|
# Get the complete conversation history.
|
|
history = [
|
|
message_item.model_dump()
|
|
for message_item in conversation.context.retrieve_full_messages()
|
|
]
|
|
total_messages = len(history)
|
|
|
|
# If a limit is specified, return only the most recent messages.
|
|
if limit and limit > 0:
|
|
history = history[-limit:]
|
|
|
|
return {
|
|
"conversation_id": conversation_id,
|
|
"messages": history,
|
|
"total_messages": total_messages,
|
|
"has_more": total_messages > len(history) if limit else False,
|
|
}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to access conversation history: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to get conversation history: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{conversation_id}/sketchpad",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Get the SketchPad content for the specified conversation",
|
|
)
|
|
async def get_conversation_sketchpad(
|
|
conversation_id: str, state: ServerState = Depends(get_server_state)
|
|
):
|
|
"""Get the SketchPad content for the specified conversation."""
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
conversation = state.conversation_manager.get_conversation(conversation_id)
|
|
if conversation is None:
|
|
return create_error_response(
|
|
message=f"Conversation {conversation_id} not found",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
# Get SketchPad content.
|
|
with conversation:
|
|
try:
|
|
# Get sketch items that include concrete content.
|
|
sketch_items = conversation.sketch_pad.list_items(include_value=True)
|
|
sketch_stats = conversation.sketch_pad.get_statistics()
|
|
|
|
return {
|
|
"conversation_id": conversation_id,
|
|
"sketch_items": sketch_items,
|
|
"statistics": sketch_stats,
|
|
"total_items": len(sketch_items),
|
|
}
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to access SketchPad: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to get SketchPad: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{conversation_id}/artifacts/latest",
|
|
response_model=Dict[str, Any],
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Get the latest generated code and model preview information for the specified conversation",
|
|
)
|
|
async def get_latest_conversation_artifacts(
|
|
conversation_id: str,
|
|
state: ServerState = Depends(get_server_state),
|
|
):
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
conversation = state.conversation_manager.get_conversation(conversation_id)
|
|
if conversation is None:
|
|
return create_error_response(
|
|
message=f"Conversation {conversation_id} not found",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
with conversation:
|
|
history = [
|
|
message_item.model_dump()
|
|
for message_item in conversation.context.retrieve_full_messages()
|
|
]
|
|
|
|
artifacts = extract_latest_artifacts(history)
|
|
code_path = artifacts.get("code_path")
|
|
code_paths = artifacts.get("code_paths", [])
|
|
model_path = artifacts.get("model_path")
|
|
model_paths = artifacts.get("model_paths", [])
|
|
output_paths = artifacts.get("output_paths", [])
|
|
|
|
response: Dict[str, Any] = {
|
|
"conversation_id": conversation_id,
|
|
"code_file": None,
|
|
"code_files": [],
|
|
"model_file": None,
|
|
"model_files": [],
|
|
"output_files": [str(path) for path in output_paths],
|
|
}
|
|
|
|
if isinstance(code_path, Path) and code_path.is_file():
|
|
response["code_file"] = _serialize_artifact_file(
|
|
conversation_id,
|
|
code_path,
|
|
include_content=True,
|
|
)
|
|
|
|
response["code_files"] = [
|
|
_serialize_artifact_file(
|
|
conversation_id,
|
|
candidate_path,
|
|
include_content=True,
|
|
)
|
|
for candidate_path in code_paths
|
|
if isinstance(candidate_path, Path) and candidate_path.is_file()
|
|
]
|
|
|
|
if isinstance(model_path, Path) and model_path.is_file():
|
|
response["model_file"] = _serialize_artifact_file(
|
|
conversation_id,
|
|
model_path,
|
|
)
|
|
|
|
response["model_files"] = [
|
|
_serialize_artifact_file(conversation_id, candidate_path)
|
|
for candidate_path in model_paths
|
|
if isinstance(candidate_path, Path) and candidate_path.is_file()
|
|
]
|
|
|
|
return response
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to get latest artifacts: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{conversation_id}/artifacts/raw",
|
|
dependencies=[Depends(get_server_state)],
|
|
description="Get the raw content of an artifact file for the specified conversation",
|
|
)
|
|
async def get_conversation_artifact_file(
|
|
conversation_id: str,
|
|
path: str,
|
|
state: ServerState = Depends(get_server_state),
|
|
):
|
|
if not state.conversation_manager:
|
|
return create_error_response(
|
|
message="Conversation manager not initialized",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|
|
|
|
try:
|
|
conversation = state.conversation_manager.get_conversation(conversation_id)
|
|
if conversation is None:
|
|
return create_error_response(
|
|
message=f"Conversation {conversation_id} not found",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
resolved_path = resolve_project_path(path)
|
|
if resolved_path is None or not resolved_path.is_file():
|
|
return create_error_response(
|
|
message="Artifact file not found or outside project root",
|
|
error_type="not_found",
|
|
status_code=404,
|
|
)
|
|
|
|
return FileResponse(
|
|
path=str(resolved_path),
|
|
media_type=guess_content_type(resolved_path),
|
|
filename=resolved_path.name,
|
|
)
|
|
except Exception as e:
|
|
return create_error_response(
|
|
message=f"Failed to get artifact file: {str(e)}",
|
|
error_type="server_error",
|
|
status_code=500,
|
|
)
|