446 lines
17 KiB
Python
446 lines
17 KiB
Python
"""Project CRUD and project-owned conversations (mode chosen once)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import hashlib
|
|
from typing import Annotated, Any, Literal
|
|
from uuid import uuid4
|
|
|
|
from fastapi import APIRouter, HTTPException, Request
|
|
from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator
|
|
|
|
from app.gateway.analysis_projects import actor, owned_project, project_store, validate_knowledge, validate_project_agent
|
|
from app.gateway.deps import get_checkpointer, get_thread_store
|
|
from app.gateway.routers.threads import ThreadCreateRequest, create_thread
|
|
from deerflow.agents.analysis_project import SOLUTION_OUTLINE, SOLUTION_TEMPLATE_ID
|
|
|
|
router = APIRouter(prefix="/api/analysis-projects", tags=["analysis-projects"])
|
|
KnowledgeId = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=128)]
|
|
DEFAULT_STRATEGY_LENSES = ["增长视角", "风险视角", "竞争视角", "资源视角", "长期价值视角"]
|
|
MemoryKind = Literal[
|
|
"fact",
|
|
"decision",
|
|
"constraint",
|
|
"preference",
|
|
"open_item",
|
|
"term",
|
|
"episode",
|
|
]
|
|
|
|
|
|
class ProjectReportConfig(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
custom_outline: str = Field(default="", max_length=60000)
|
|
structure_mode: Literal["fixed", "adaptive"] = "fixed"
|
|
retrieval_directions: list[Annotated[str, StringConstraints(max_length=200)]] = Field(default_factory=list, max_length=12)
|
|
retrieval_skills: list[Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=64, pattern=r"^\S+$")]] = Field(default_factory=list, max_length=8)
|
|
|
|
|
|
class ProjectStrategyConfig(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
lenses: list[Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=40)]] = Field(default_factory=lambda: list(DEFAULT_STRATEGY_LENSES), min_length=1, max_length=10)
|
|
|
|
@field_validator("lenses")
|
|
@classmethod
|
|
def unique_lenses(cls, value: list[str]) -> list[str]:
|
|
return list(dict.fromkeys(value))
|
|
|
|
|
|
class ProjectFields(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
name: str = Field(min_length=1, max_length=160)
|
|
background: str = Field(default="", max_length=12000)
|
|
agent_id: KnowledgeId | None = None
|
|
report_config: ProjectReportConfig = Field(default_factory=ProjectReportConfig)
|
|
strategy_config: ProjectStrategyConfig = Field(default_factory=ProjectStrategyConfig)
|
|
knowledge_base_ids: list[KnowledgeId] = Field(default_factory=list, max_length=30)
|
|
assistant_knowledge_base_ids: list[KnowledgeId] = Field(default_factory=list, max_length=30)
|
|
|
|
@field_validator("knowledge_base_ids", "assistant_knowledge_base_ids")
|
|
@classmethod
|
|
def unique_ids(cls, value: list[str]) -> list[str]:
|
|
return list(dict.fromkeys(value))
|
|
|
|
|
|
class ProjectCreate(ProjectFields):
|
|
mode: Literal["qa", "strategy", "solution"]
|
|
|
|
|
|
def public_project(project: dict) -> dict:
|
|
return {
|
|
**{key: value for key, value in project.items() if key != "user_id"},
|
|
"solution_template": {
|
|
"id": SOLUTION_TEMPLATE_ID,
|
|
"name": "当前项目报告结构",
|
|
"outline": (project.get("report_config") or {}).get("custom_outline") or SOLUTION_OUTLINE,
|
|
}
|
|
if project["mode"] == "solution"
|
|
else None,
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_projects(request: Request):
|
|
user_id, _ = await actor(request)
|
|
return {"projects": [public_project(p) for p in await project_store(request).list(user_id)]}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def new_project(body: ProjectCreate, request: Request):
|
|
user_id, _ = await actor(request)
|
|
await validate_knowledge(request, body.model_dump())
|
|
await validate_project_agent(request, body.agent_id)
|
|
return public_project(await project_store(request).create(user_id, **body.model_dump()))
|
|
|
|
|
|
@router.get("/{project_id}")
|
|
async def get_project(project_id: str, request: Request):
|
|
return public_project(await owned_project(request, project_id))
|
|
|
|
|
|
@router.put("/{project_id}")
|
|
async def update_project(project_id: str, body: ProjectFields, request: Request):
|
|
project = await owned_project(request, project_id)
|
|
await validate_knowledge(request, {**project, **body.model_dump()})
|
|
await validate_project_agent(request, body.agent_id)
|
|
updated = await project_store(request).update(project_id, project["user_id"], **body.model_dump())
|
|
if updated is None:
|
|
raise HTTPException(404, "项目不存在")
|
|
return public_project(updated)
|
|
|
|
|
|
def _project_memory_store(request: Request):
|
|
store = getattr(request.app.state, "analysis_project_memory_store", None)
|
|
if store is None:
|
|
raise HTTPException(503, "项目记忆存储尚未就绪")
|
|
return store
|
|
|
|
|
|
async def _qa_project(request: Request, project_id: str) -> dict:
|
|
project = await owned_project(request, project_id)
|
|
if project.get("mode") != "qa":
|
|
raise HTTPException(422, "跨会话项目记忆仅适用于普通问答项目")
|
|
return project
|
|
|
|
|
|
async def _project_memory_view(
|
|
request: Request,
|
|
project: dict,
|
|
*,
|
|
item_limit: int = 100,
|
|
item_offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
store = _project_memory_store(request)
|
|
state = await store.get_state(project["id"], project["user_id"], create=True)
|
|
state = state or {}
|
|
settings_view = await store.get_settings(project["id"], project["user_id"])
|
|
nested_settings = (
|
|
settings_view.get("settings")
|
|
if isinstance(settings_view, dict)
|
|
and isinstance(settings_view.get("settings"), dict)
|
|
else {}
|
|
)
|
|
if not nested_settings and isinstance(settings_view, dict):
|
|
nested_settings = {
|
|
key: value for key, value in settings_view.items() if key != "enabled"
|
|
}
|
|
enabled = bool(
|
|
settings_view.get("enabled", state.get("enabled", True))
|
|
if isinstance(settings_view, dict)
|
|
else state.get("enabled", True)
|
|
)
|
|
items = await store.list_items(
|
|
project["id"],
|
|
project["user_id"],
|
|
status="active",
|
|
limit=item_limit,
|
|
offset=item_offset,
|
|
)
|
|
snapshot = state.get("snapshot") if isinstance(state.get("snapshot"), dict) else {}
|
|
item_count = int(state.get("item_count") or len(items))
|
|
status = str(state.get("status") or "ready")
|
|
if status == "ready" and item_count == 0 and not snapshot:
|
|
status = "empty"
|
|
return {
|
|
"project_id": project["id"],
|
|
"enabled": enabled,
|
|
"effective_enabled": enabled,
|
|
# Legacy compatibility aliases now describe project-memory subsystem
|
|
# readiness. The ordinary/personal-memory global switch is unrelated.
|
|
"global_enabled": True,
|
|
"available": True,
|
|
"runtime_enabled": enabled,
|
|
"settings": nested_settings,
|
|
"status": status,
|
|
"revision": int(state.get("revision") or 0),
|
|
"item_count": item_count,
|
|
"last_error": state.get("last_error"),
|
|
"last_built_at": state.get("rebuilt_at") or state.get("last_built_at"),
|
|
"rebuilt_at": state.get("rebuilt_at") or state.get("last_built_at"),
|
|
"updated_at": state.get("updated_at"),
|
|
"snapshot": snapshot,
|
|
"summary": str(snapshot.get("summary") or ""),
|
|
"items": [_public_memory_item(item) for item in items],
|
|
"limit": item_limit,
|
|
"offset": item_offset,
|
|
}
|
|
|
|
|
|
def _public_memory_item(item: dict[str, Any]) -> dict[str, Any]:
|
|
"""Keep 1024-d vectors out of ordinary management responses."""
|
|
|
|
embedding = item.get("embedding")
|
|
return {
|
|
key: value
|
|
for key, value in item.items()
|
|
if key not in {"embedding", "embedding_json"}
|
|
} | {"has_embedding": bool(embedding)}
|
|
|
|
|
|
class ProjectMemoryTuning(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
item_limit: int | None = Field(default=None, ge=1, le=20)
|
|
episode_limit: int | None = Field(default=None, ge=0, le=5)
|
|
token_budget: int | None = Field(default=None, ge=400, le=4000)
|
|
|
|
|
|
class ProjectMemorySettingsUpdate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
enabled: bool | None = None
|
|
settings: ProjectMemoryTuning | None = None
|
|
|
|
|
|
class ProjectMemoryItemCreate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
logical_key: str | None = Field(default=None, max_length=180)
|
|
kind: MemoryKind
|
|
content: str = Field(min_length=1, max_length=2000)
|
|
confidence: float = Field(default=1.0, ge=0, le=1)
|
|
importance: float = Field(default=0.7, ge=0, le=1)
|
|
pinned: bool = False
|
|
|
|
|
|
class ProjectMemoryItemUpdate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
|
|
logical_key: str | None = Field(default=None, max_length=180)
|
|
kind: MemoryKind | None = None
|
|
content: str | None = Field(default=None, min_length=1, max_length=2000)
|
|
confidence: float | None = Field(default=None, ge=0, le=1)
|
|
importance: float | None = Field(default=None, ge=0, le=1)
|
|
pinned: bool | None = None
|
|
|
|
|
|
@router.get("/{project_id}/memory")
|
|
async def get_project_memory(
|
|
project_id: str,
|
|
request: Request,
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
):
|
|
project = await _qa_project(request, project_id)
|
|
return await _project_memory_view(
|
|
request,
|
|
project,
|
|
item_limit=max(1, min(200, limit)),
|
|
item_offset=max(0, offset),
|
|
)
|
|
|
|
|
|
@router.patch("/{project_id}/memory/settings")
|
|
async def update_project_memory_settings(
|
|
project_id: str,
|
|
body: ProjectMemorySettingsUpdate,
|
|
request: Request,
|
|
):
|
|
project = await _qa_project(request, project_id)
|
|
if body.enabled is None and body.settings is None:
|
|
raise HTTPException(422, "请提供要更新的记忆设置")
|
|
settings = (
|
|
body.settings.model_dump(exclude_none=True) if body.settings is not None else None
|
|
)
|
|
await _project_memory_store(request).set_settings(
|
|
project["id"],
|
|
project["user_id"],
|
|
enabled=body.enabled,
|
|
settings=settings,
|
|
)
|
|
return await _project_memory_view(request, project)
|
|
|
|
|
|
@router.post("/{project_id}/memory/rebuild", status_code=202)
|
|
async def rebuild_project_memory(project_id: str, request: Request):
|
|
from app.gateway.analysis_project_memory import enqueue_project_rebuild
|
|
|
|
project = await _qa_project(request, project_id)
|
|
settings = await _project_memory_store(request).get_settings(
|
|
project["id"], project["user_id"]
|
|
)
|
|
if not bool(settings.get("enabled", True)):
|
|
raise HTTPException(409, "当前项目的跨会话记忆已暂停")
|
|
ids = await project_store(request).thread_ids(project["id"], project["user_id"])
|
|
queued = await enqueue_project_rebuild(
|
|
app=request.app,
|
|
project=project,
|
|
thread_ids=ids,
|
|
)
|
|
return {
|
|
**await _project_memory_view(request, project),
|
|
"queued_jobs": queued,
|
|
}
|
|
|
|
|
|
@router.delete("/{project_id}/memory")
|
|
async def clear_project_memory(project_id: str, request: Request):
|
|
project = await _qa_project(request, project_id)
|
|
await _project_memory_store(request).clear(project["id"], project["user_id"])
|
|
return await _project_memory_view(request, project)
|
|
|
|
|
|
@router.post("/{project_id}/memory/items", status_code=201)
|
|
async def create_project_memory_item(
|
|
project_id: str,
|
|
body: ProjectMemoryItemCreate,
|
|
request: Request,
|
|
):
|
|
from app.gateway.analysis_project_memory import sanitize_project_memory_text
|
|
|
|
project = await _qa_project(request, project_id)
|
|
content = sanitize_project_memory_text(body.content)
|
|
logical_key = (
|
|
sanitize_project_memory_text(body.logical_key)
|
|
if body.logical_key
|
|
else "manual:"
|
|
+ hashlib.sha256(f"{body.kind}\0{content.casefold()}".encode()).hexdigest()
|
|
)
|
|
item = await _project_memory_store(request).upsert_item(
|
|
project["id"],
|
|
project["user_id"],
|
|
logical_key=logical_key,
|
|
kind=body.kind,
|
|
content=content,
|
|
confidence=body.confidence,
|
|
importance=body.importance,
|
|
pinned=body.pinned,
|
|
source_refs=[{"kind": "manual"}],
|
|
metadata={"manual": True},
|
|
)
|
|
return _public_memory_item(item)
|
|
|
|
|
|
@router.patch("/{project_id}/memory/items/{item_id}")
|
|
async def update_project_memory_item(
|
|
project_id: str,
|
|
item_id: str,
|
|
body: ProjectMemoryItemUpdate,
|
|
request: Request,
|
|
):
|
|
from app.gateway.analysis_project_memory import sanitize_project_memory_text
|
|
|
|
project = await _qa_project(request, project_id)
|
|
fields = body.model_dump(exclude_unset=True, exclude_none=True)
|
|
if "content" in fields:
|
|
fields["content"] = sanitize_project_memory_text(fields["content"])
|
|
if "logical_key" in fields:
|
|
fields["logical_key"] = sanitize_project_memory_text(fields["logical_key"])
|
|
if not fields:
|
|
raise HTTPException(422, "请提供要更新的记忆内容")
|
|
item = await _project_memory_store(request).update_item(
|
|
project["id"], project["user_id"], item_id, **fields
|
|
)
|
|
if item is None:
|
|
raise HTTPException(404, "项目记忆条目不存在")
|
|
return _public_memory_item(item)
|
|
|
|
|
|
@router.delete("/{project_id}/memory/items/{item_id}", status_code=204)
|
|
async def delete_project_memory_item(
|
|
project_id: str,
|
|
item_id: str,
|
|
request: Request,
|
|
):
|
|
project = await _qa_project(request, project_id)
|
|
deleted = await _project_memory_store(request).delete_item(
|
|
project["id"], project["user_id"], item_id
|
|
)
|
|
if not deleted:
|
|
raise HTTPException(404, "项目记忆条目不存在")
|
|
return None
|
|
|
|
|
|
@router.get("/{project_id}/threads")
|
|
async def project_threads(project_id: str, request: Request):
|
|
project = await owned_project(request, project_id)
|
|
ids = await project_store(request).thread_ids(project_id, project["user_id"])
|
|
threads = []
|
|
for thread_id in ids:
|
|
row = await get_thread_store(request).get(thread_id, user_id=project["user_id"])
|
|
if row:
|
|
metadata = row.get("metadata") or {}
|
|
if metadata.get("system") or metadata.get("analysis_project_id") != project_id or metadata.get("analysis_project_mode") != project["mode"] or metadata.get("analysis_project_name") != project["name"]:
|
|
metadata = {
|
|
**metadata,
|
|
"system": False,
|
|
"analysis_project_id": project_id,
|
|
"analysis_project_mode": project["mode"],
|
|
"analysis_project_name": project["name"],
|
|
}
|
|
await get_thread_store(request).update_metadata(thread_id, metadata, user_id=project["user_id"])
|
|
row = {**row, "metadata": metadata}
|
|
threads.append({**row, "title": row.get("display_name") or (row.get("values") or {}).get("title") or "新对话"})
|
|
threads.sort(key=lambda row: str(row.get("updated_at") or ""), reverse=True)
|
|
return {"threads": threads}
|
|
|
|
|
|
class ProjectThreadCreate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
agent_id: KnowledgeId | None = None
|
|
|
|
|
|
@router.post("/{project_id}/threads", status_code=201)
|
|
async def new_project_thread(project_id: str, request: Request, body: ProjectThreadCreate | None = None):
|
|
project = await owned_project(request, project_id)
|
|
store = project_store(request)
|
|
thread_id = str(uuid4())
|
|
agent_id = body.agent_id if body and "agent_id" in body.model_fields_set else project.get("agent_id")
|
|
await validate_project_agent(request, agent_id)
|
|
await store.bind_thread(project_id, project["user_id"], thread_id, agent_id=agent_id)
|
|
try:
|
|
# Binding is server-only; this endpoint never accepts an arbitrary
|
|
# existing thread id, so conversations cannot be moved between projects.
|
|
body = ThreadCreateRequest(
|
|
thread_id=thread_id,
|
|
assistant_id=agent_id or "lead_agent",
|
|
metadata={
|
|
"thread_type": "analysis_project",
|
|
"system": False,
|
|
"analysis_project_id": project_id,
|
|
"analysis_project_mode": project["mode"],
|
|
"analysis_project_name": project["name"],
|
|
"agent_id": agent_id,
|
|
},
|
|
)
|
|
result = await create_thread(body=body, request=request)
|
|
except BaseException:
|
|
|
|
async def cleanup():
|
|
await get_thread_store(request).delete(thread_id)
|
|
await get_checkpointer(request).adelete_thread(thread_id)
|
|
await store.unbind_failed_creation(thread_id)
|
|
|
|
await asyncio.shield(cleanup())
|
|
raise
|
|
return result
|
|
|
|
|
|
@router.get("/{project_id}/threads/{thread_id}")
|
|
async def get_project_thread(project_id: str, thread_id: str, request: Request):
|
|
project = await owned_project(request, project_id)
|
|
bound = await project_store(request).project_for_thread(thread_id)
|
|
if not bound or bound["id"] != project["id"]:
|
|
raise HTTPException(404, "该对话不属于当前项目")
|
|
row = await get_thread_store(request).get(thread_id, user_id=project["user_id"])
|
|
if not row:
|
|
raise HTTPException(404, "对话不存在")
|
|
return row
|