chore: retire legacy wolai backend runtime

This commit is contained in:
lix-2026
2026-05-28 22:02:10 +08:00
parent 39b9a0183a
commit 1109e3c0d8
31 changed files with 21 additions and 5785 deletions
Binary file not shown.
+9 -25
View File
@@ -1,10 +1,12 @@
# wolai-backendstage0 骨架)
# wolai-backend
本目录是 design3.0 指南下的 FastAPI + Celery 阶段 0 脚手架。当前目标:
本目录现在只保留轻量 FastAPI 辅助后端。OpenAI Agents 文档代理、Celery/Redis 队列和旧 OCR 占位链路已退役到仓库 `recycle/`,当前 AI 主链统一走 Hermes / Reasonix。
1. 暴露健康检查与占位 API`/api/v1/tasks/ocr``/api/v1/chat`),供前端联调。
2. 预留 Celery 队列、LightRAG、MinerU 集成点,但将重计算逻辑延后到阶段 1。
3. 依赖全部维护在 `requirements.txt`,其中 LightRAG 与 MinerU 通过本地源码安装,使用项目根目录已经下载好的仓库。
当前保留:
1. 健康检查:`GET /health`
2. CORS 配置:读取仓库根目录 `.env.all` 与环境变量
3. 依赖列表:`requirements.txt`
## 环境初始化
@@ -18,7 +20,7 @@ python -m venv venv
# set ALL_PROXY=
# set NO_PROXY=*
# 安装 PyPI 依赖 + 本地 LightRAG (项目根目录已放置源码)
# 安装 PyPI 依赖
pip install -r requirements.txt
```
@@ -29,31 +31,13 @@ pip install -r requirements.txt
```powershell
# FastAPI
uvicorn app.main:app --reload --port 8000
# Celery worker(后续阶段会真正处理 MinerU / LightRAG 作业)
celery -A app.workers.celery_app worker --loglevel=info
```
## 目录速览
```
app/
├── config.py # Pydantic 设置,统一读取 .env
├── deps.py # Supabase JWT / 客户端依赖
├── config.py # Pydantic 设置,统一读取仓库根目录 .env.all
├── routers/
│ ├── health.py # GET /health
│ ├── tasks.py # POST /api/v1/tasks/ocr、GET /api/v1/tasks/{id}
│ └── chat.py # GET /api/v1/chatSSE 占位)
├── schemas/ # Pydantic 请求/响应模型
├── services/ # LightRAG、MinerU、Storage 等服务占位
└── workers/
├── celery_app.py # Celery 配置
└── tasks.py # 后台任务入口(当前返回占位日志)
```
阶段 0 完成后,stage1 将在此基础上补充:
- MinerU OCR 实装(下载 Supabase Storage 文件 → 解析 → 更新 `documents`
- LightRAG 的增量索引 / 查询
- SSE 流式回答 + 来源引用
- Redis/APScheduler 任务状态回传
-11
View File
@@ -15,18 +15,7 @@ class Settings(BaseSettings):
env_file=str(ENV_ALL_PATH), env_file_encoding="utf-8", extra="ignore"
)
# 说明:项目迁移到 Convex 的过程中,后端可能暂时不依赖 Supabase。
# 因此这里给出空字符串默认值,避免本地未配置 .env 时无法启动开发服务器。
supabase_url: str = ""
supabase_service_role_key: str = ""
redis_url: str = "redis://localhost:6379/0"
frontend_url: str = "http://localhost:3000"
openai_api_key: str = ""
mnote_ai_default_model: str = ""
mnote_ai_orchestrator_api_key: str = ""
openai_agents_session_db_path: str = ""
lightrag_db_url: str = ""
lightrag_collection: str = "wolai-docs"
@lru_cache
-58
View File
@@ -1,58 +0,0 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
try: # Python 3.9+
from typing import Annotated # type: ignore[attr-defined]
except ImportError: # Python 3.8 fallback
from typing_extensions import Annotated
import httpx
from fastapi import Depends, Header, HTTPException, status
from app.config import settings
@dataclass
class AuthContext:
user_id: str
access_token: str
async def get_current_user(
authorization: Annotated[Optional[str], Header(convert_underscores=False)] = None,
) -> AuthContext:
"""
验证 Supabase JWTstage0 直接依赖 service_role 解析 token。
生产环境应通过 API Gateway 注入 user。
"""
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")
token = authorization.replace("Bearer ", "", 1).strip()
if not token:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Empty token")
auth_url = f"{settings.supabase_url.rstrip('/')}/auth/v1/user"
headers = {
"Authorization": f"Bearer {token}",
"apikey": settings.supabase_service_role_key,
}
async with httpx.AsyncClient(timeout=10.0) as client:
try:
response = await client.get(auth_url, headers=headers)
response.raise_for_status()
except httpx.HTTPError as exc: # pragma: no cover - 网络异常
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Supabase token") from exc
data = response.json()
user_id = data.get("id") or data.get("user", {}).get("id")
if not user_id:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Supabase token")
return AuthContext(user_id=str(user_id), access_token=token)
AuthDep = Annotated[AuthContext, Depends(get_current_user)]
+1 -5
View File
@@ -1,12 +1,8 @@
from fastapi import APIRouter
from . import ai_agent, chat, health, luckysheet_ws, tasks
from . import health
api_router = APIRouter(prefix="/api/v1")
api_router.include_router(tasks.router, tags=["tasks"])
api_router.include_router(chat.router, tags=["chat"])
api_router.include_router(ai_agent.router, tags=["ai-agent"])
root_router = APIRouter()
root_router.include_router(health.router, tags=["health"])
root_router.include_router(luckysheet_ws.router)
-64
View File
@@ -1,64 +0,0 @@
from __future__ import annotations
from typing import Annotated
from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from fastapi.responses import JSONResponse, StreamingResponse
from app.config import settings
from app.services.ai_document_agent import (
DocumentAiRunRequest,
build_document_ai_config_payload,
has_configured_openai_api_key,
run_document_agent_stream,
sdk_available,
)
router = APIRouter(prefix="/ai-agent")
def verify_internal_key(
x_mnote_ai_key: Annotated[str | None, Header()] = None,
) -> None:
expected = settings.mnote_ai_orchestrator_api_key.strip()
if not expected:
return
if x_mnote_ai_key == expected:
return
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid orchestrator key",
)
@router.get("/health")
async def health() -> JSONResponse:
return JSONResponse(
{
"ok": True,
"bridge": "openai_agents_python",
"sdkAvailable": sdk_available(),
"openaiConfigured": has_configured_openai_api_key(),
"bridgeRuntimeMode": "local_runtime",
}
)
@router.post("/document/run")
async def run_document(
request: Request,
payload: DocumentAiRunRequest,
_: None = Depends(verify_internal_key),
) -> StreamingResponse:
stream = run_document_agent_stream(
payload,
source_headers={key.lower(): value for key, value in request.headers.items()},
)
return StreamingResponse(stream, media_type="text/event-stream")
@router.get("/document/config")
async def get_document_config(
_: None = Depends(verify_internal_key),
) -> JSONResponse:
return JSONResponse(build_document_ai_config_payload())
-26
View File
@@ -1,26 +0,0 @@
import asyncio
from fastapi import APIRouter
from fastapi.responses import StreamingResponse
from app.deps import AuthDep
from app.services.lightrag_service import lightrag_service
router = APIRouter(prefix="/chat")
@router.get("")
async def chat(query: str, document_id: str, auth: AuthDep) -> StreamingResponse: # noqa: ARG001
"""
SSE 流式占位。后续会调用 LightRAG + OpenAI。
当前直接返回 mock 文字,确保前端链路可用。
"""
async def event_stream() -> asyncio.AsyncGenerator[str, None]:
reply = await lightrag_service.query(query_text=query, user_id="placeholder-user")
chunks = [reply[: len(reply) // 2 or 1], reply[len(reply) // 2 or 1 :]]
for chunk in chunks:
yield f"data: {chunk}\n\n"
await asyncio.sleep(0.1)
yield "data: [DONE]\n\n"
return StreamingResponse(event_stream(), media_type="text/event-stream")
-195
View File
@@ -1,195 +0,0 @@
from __future__ import annotations
import asyncio
import json
import logging
import urllib.parse
import zlib
from dataclasses import dataclass
from typing import Dict, List, Optional
import httpx
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, status
from starlette.concurrency import run_in_threadpool
from app.config import settings
from app.services.supabase_rest import supabase_rest
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/ws", tags=["luckysheet"])
@dataclass
class LuckysheetClient:
websocket: WebSocket
user_id: str
username: str
class LuckysheetConnectionManager:
def __init__(self) -> None:
self._clients: Dict[str, List[LuckysheetClient]] = {}
self._lock = asyncio.Lock()
async def add(self, grid_key: str, client: LuckysheetClient) -> None:
async with self._lock:
self._clients.setdefault(grid_key, []).append(client)
logger.debug("Luckysheet client %s joined grid %s", client.user_id, grid_key)
async def remove(self, grid_key: str, websocket: WebSocket) -> None:
async with self._lock:
clients = self._clients.get(grid_key)
if not clients:
return
self._clients[grid_key] = [client for client in clients if client.websocket is not websocket]
if not self._clients[grid_key]:
self._clients.pop(grid_key, None)
logger.debug("Luckysheet connection removed from grid %s", grid_key)
async def broadcast_payload(
self,
grid_key: str,
sender: LuckysheetClient,
payload: str,
event_type: int,
) -> None:
clients = self._clients.get(grid_key, [])
if not clients:
return
message = json.dumps(
{
"data": payload,
"id": sender.user_id,
"username": sender.username,
"type": event_type,
},
ensure_ascii=False,
)
for client in clients:
if client.websocket is sender.websocket:
continue
try:
await client.websocket.send_text(message)
except Exception as exc:
logger.warning("Failed to forward message to %s: %s", client.user_id, exc)
async def broadcast_exit(self, grid_key: str, user_id: str) -> None:
clients = self._clients.get(grid_key, [])
if not clients:
return
message = json.dumps({"message": "用户退出", "id": user_id}, ensure_ascii=False)
for client in clients:
try:
await client.websocket.send_text(message)
except Exception as exc:
logger.warning("Failed to notify client exit: %s", exc)
manager = LuckysheetConnectionManager()
def _decode_ws_payload(raw_message: str) -> Optional[dict]:
try:
compressed = raw_message.encode("latin1")
inflated = zlib.decompress(compressed)
decoded = urllib.parse.unquote(inflated.decode("utf-8"))
return json.loads(decoded)
except Exception as exc:
logger.debug("Failed to decode luckysheet payload: %s", exc)
return None
async def _fetch_supabase_user(access_token: str) -> Optional[dict]:
base_url = settings.supabase_url.rstrip("/")
headers = {
"apikey": settings.supabase_service_role_key,
"Authorization": f"Bearer {access_token}",
}
try:
async with httpx.AsyncClient(timeout=5.0) as client:
response = await client.get(f"{base_url}/auth/v1/user", headers=headers)
response.raise_for_status()
return response.json()
except httpx.HTTPError as exc:
logger.warning("Supabase auth verification failed: %s", exc)
return None
async def _fetch_table_by_grid_key(grid_key: str) -> Optional[dict]:
try:
return await run_in_threadpool(lambda: supabase_rest.select_one("document_tables", {"grid_key": grid_key}))
except Exception as exc:
logger.error("Failed to query document_tables: %s", exc)
return None
async def _is_workspace_member(workspace_id: str, user_id: str) -> bool:
try:
result = await run_in_threadpool(
lambda: supabase_rest.select_one("workspace_members", {"workspace_id": workspace_id, "user_id": user_id}),
)
return bool(result)
except Exception as exc:
logger.error("Failed to verify workspace membership: %s", exc)
return False
@router.websocket("/luckysheet")
async def luckysheet_collaboration(websocket: WebSocket) -> None:
params = websocket.query_params
grid_key = params.get("gridKey")
token = params.get("token")
raw_user_id = params.get("userid") or params.get("userId")
requested_type = params.get("type") or "luckysheet"
username = params.get("username") or ""
if requested_type != "luckysheet" or not grid_key or not token or not raw_user_id:
await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="缺少协同参数")
return
supabase_user = await _fetch_supabase_user(token)
if not supabase_user or supabase_user.get("id") != raw_user_id:
await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="身份验证失败")
return
table_row = await _fetch_table_by_grid_key(grid_key)
if not table_row or not table_row.get("workspace_id"):
await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="gridKey 无效")
return
workspace_id = table_row["workspace_id"]
if not await _is_workspace_member(workspace_id, raw_user_id):
await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="无权访问该表格")
return
fallback_name = (
username
or supabase_user.get("user_metadata", {}).get("full_name")
or supabase_user.get("email")
or f"用户-{raw_user_id[:6]}"
)
client = LuckysheetClient(websocket=websocket, user_id=raw_user_id, username=fallback_name)
await websocket.accept()
await manager.add(grid_key, client)
logger.info("Luckysheet client %s connected to %s", raw_user_id, grid_key)
try:
while True:
message = await websocket.receive_text()
if message == "rub":
continue
decoded = _decode_ws_payload(message)
if not decoded:
continue
op_type = decoded.get("t")
event_type = 3 if op_type == "mv" else 2
await manager.broadcast_payload(grid_key, client, message, event_type)
except WebSocketDisconnect:
logger.info("Luckysheet client %s disconnected", raw_user_id)
except Exception as exc:
logger.error("Luckysheet websocket error: %s", exc)
finally:
await manager.remove(grid_key, websocket)
await manager.broadcast_exit(grid_key, raw_user_id)
-29
View File
@@ -1,29 +0,0 @@
from fastapi import APIRouter, HTTPException, status
from app.deps import AuthDep
from app.schemas.tasks import OcrTaskRequest, TaskStatusResponse
from app.services.task_tracker import task_tracker
from app.workers.tasks import ocr_pipeline
router = APIRouter(prefix="/tasks")
@router.post("/ocr", response_model=TaskStatusResponse)
async def enqueue_ocr_task(payload: OcrTaskRequest, auth: AuthDep) -> TaskStatusResponse:
"""记录任务并投递 Celery,阶段 0 返回占位任务。"""
task = task_tracker.create_task(user_id=auth.user_id, document_id=payload.document_id, task_type="ocr")
ocr_pipeline.delay(
task_id=task.task_id,
document_id=payload.document_id,
file_url=str(payload.file_url),
user_id=auth.user_id,
)
return task
@router.get("/{task_id}", response_model=TaskStatusResponse)
async def get_task_status(task_id: str, auth: AuthDep) -> TaskStatusResponse:
task = task_tracker.get_task(task_id=task_id, user_id=auth.user_id)
if not task:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Task not found")
return task
-17
View File
@@ -1,17 +0,0 @@
from __future__ import annotations
from typing import Literal, Optional
from pydantic import AnyHttpUrl, BaseModel
class OcrTaskRequest(BaseModel):
document_id: str
file_url: AnyHttpUrl
class TaskStatusResponse(BaseModel):
task_id: str
status: Literal["pending", "processing", "completed", "failed"]
progress: int = 0
message: Optional[str] = None
File diff suppressed because it is too large Load Diff
@@ -1,19 +0,0 @@
"""LightRAG 集成占位。阶段 1 会在这里封装真正的查询与索引。"""
from app.config import settings
class LightRAGService:
def __init__(self) -> None:
self.collection = settings.lightrag_collection
async def queue_index(self, document_id: str, raw_text: str) -> None:
"""预留方法:后续调用 LightRAG.update_index。"""
return None
async def query(self, query_text: str, user_id: str) -> str:
"""预留方法:后续调用 LightRAG.query,当前返回占位回答。"""
return f"[mock] {query_text}"
lightrag_service = LightRAGService()
@@ -1,13 +0,0 @@
"""MinerU OCR 占位服务。"""
class MinerUService:
async def extract_markdown(self, file_url: str) -> str:
"""
阶段 0:直接返回固定内容,保证前端流程贯通。
阶段 1:调用 MinerU CLI / SDK,从 Supabase Storage 下载文件后解析。
"""
return f"# OCR Placeholder\n\n源文件:{file_url}"
mineru_service = MinerUService()
@@ -1,57 +0,0 @@
from __future__ import annotations
from typing import Any, Dict, Optional
import httpx
from app.config import settings
class SupabaseRestClient:
"""轻量封装 Supabase RESTful API,兼容本地 sb_secret 密钥。"""
def __init__(self) -> None:
base_url = settings.supabase_url.rstrip("/")
self.client = httpx.Client(
base_url=f"{base_url}/rest/v1",
headers={
"apikey": settings.supabase_service_role_key,
"Authorization": f"Bearer {settings.supabase_service_role_key}",
},
timeout=10.0,
)
def insert(self, table: str, payload: Dict[str, Any]) -> Dict[str, Any]:
response = self.client.post(
f"/{table}",
json=payload,
headers={"Prefer": "return=representation"},
)
response.raise_for_status()
data = response.json()
if isinstance(data, list):
return data[0]
return data
def select_one(self, table: str, filters: Dict[str, Any]) -> Optional[Dict[str, Any]]:
params = {key: f"eq.{value}" for key, value in filters.items()}
params["select"] = "*"
response = self.client.get(f"/{table}", params=params)
response.raise_for_status()
data = response.json()
if isinstance(data, list) and data:
return data[0]
return None
def update(self, table: str, filters: Dict[str, Any], payload: Dict[str, Any]) -> None:
params = {key: f"eq.{value}" for key, value in filters.items()}
response = self.client.patch(
f"/{table}",
params=params,
json=payload,
headers={"Prefer": "return=minimal"},
)
response.raise_for_status()
supabase_rest = SupabaseRestClient()
@@ -1,43 +0,0 @@
from __future__ import annotations
from typing import Any, Dict, Optional
from app.schemas.tasks import TaskStatusResponse
from app.services.supabase_rest import supabase_rest
class SupabaseTaskTracker:
"""利用 Supabase background_tasks 表追踪进度。"""
def create_task(self, *, user_id: str, document_id: str, task_type: str = "ocr") -> TaskStatusResponse:
payload = {
"user_id": user_id,
"document_id": document_id,
"task_type": task_type,
"status": "pending",
"progress": 0,
}
record = supabase_rest.insert("background_tasks", payload)
return self._to_response(record)
def get_task(self, *, task_id: str, user_id: str) -> Optional[TaskStatusResponse]:
record = supabase_rest.select_one("background_tasks", {"id": task_id, "user_id": user_id})
if not record:
return None
return self._to_response(record)
def update_task(self, *, task_id: str, **kwargs: Any) -> None:
supabase_rest.update("background_tasks", {"id": task_id}, kwargs)
def _to_response(self, record: Optional[Dict[str, Any]]) -> TaskStatusResponse:
if not record:
raise ValueError("Task record missing")
return TaskStatusResponse(
task_id=str(record["id"]),
status=record.get("status", "pending"),
progress=record.get("progress", 0),
message=record.get("message"),
)
task_tracker = SupabaseTaskTracker()
-15
View File
@@ -1,15 +0,0 @@
from celery import Celery
from app.config import settings
def create_celery_app() -> Celery:
app = Celery("wolai-backend")
app.conf.broker_url = settings.redis_url
app.conf.result_backend = settings.redis_url
app.conf.task_routes = {"app.workers.tasks.*": {"queue": "wolai-tasks"}}
app.autodiscover_tasks(["app.workers"])
return app
celery_app = create_celery_app()
-51
View File
@@ -1,51 +0,0 @@
from __future__ import annotations
from typing import Dict
from celery import shared_task
from app.services.supabase_rest import supabase_rest
@shared_task(name="app.workers.tasks.ocr_pipeline")
def ocr_pipeline(task_id: str, document_id: str, file_url: str, user_id: str) -> Dict[str, str]:
"""
阶段 0 Celery 任务:模拟 OCR,向 documents+background_tasks 回写占位结果。
阶段 1 将在此处串联 MinerU OCR 与 LightRAG 索引。
"""
supabase_rest.update("background_tasks", {"id": task_id}, {"status": "processing", "progress": 30})
markdown = f"# OCR 结果占位\\n\\n文件地址:{file_url}\\n\\n> 阶段 1 将替换为 MinerU 输出。"
supabase_rest.update(
"documents",
{"id": document_id, "user_id": user_id},
{
"content": {
"blocks": [
{
"type": "paragraph",
"text": markdown,
}
]
},
"raw_text": markdown,
"index_status": "completed",
},
)
supabase_rest.update(
"background_tasks",
{"id": task_id},
{
"status": "completed",
"progress": 100,
"message": "OCR 模拟完成",
},
)
return {
"task_id": task_id,
"document_id": document_id,
"file_url": file_url,
"status": "completed",
}
-17
View File
@@ -1,17 +0,0 @@
INFO: Will watch for changes in these directories: ['/mnt/Data1T/mnote/wolai-backend']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [1431406] using WatchFiles
INFO: Started server process [1431452]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: 127.0.0.1:47792 - "HEAD / HTTP/1.1" 405 Method Not Allowed
INFO: 127.0.0.1:37784 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:36884 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:48348 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:53220 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:48870 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:49756 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:49756 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:49732 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:49732 - "GET /health HTTP/1.1" 200 OK
INFO: 127.0.0.1:49732 - "GET /health HTTP/1.1" 200 OK
View File
File diff suppressed because it is too large Load Diff
-19
View File
@@ -1,19 +0,0 @@
-------------- celery@LIX v5.4.0 (opalescent)
--- ***** -----
-- ******* ---- Windows-10-10.0.26100-SP0 2025-11-18 18:34:36
- *** --- * ---
- ** ---------- [config]
- ** ---------- .> app: wolai-backend:0x2b6d01c47d0
- ** ---------- .> transport: redis://localhost:6379/0
- ** ---------- .> results: redis://localhost:6379/0
- *** --- * --- .> concurrency: 24 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** -----
-------------- [queues]
.> celery exchange=celery(direct) key=celery
[tasks]
. app.workers.tasks.ocr_pipeline
-8
View File
@@ -1,12 +1,4 @@
fastapi==0.115.5
uvicorn[standard]==0.33.0
celery[redis]==5.4.0
redis==5.1.1
httpx==0.27.2
python-dotenv==1.0.1
pgvector==0.3.5
openai==2.11.0
openai-agents==0.7.0
python-multipart==0.0.17
pydantic-settings==2.6.1
typing_extensions==4.15.0
@@ -1,362 +0,0 @@
import contextlib
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
from app.services.ai_document_agent import (
DocumentAiEmitter,
DocumentAiContextPayload,
DocumentAiRunContext,
DocumentAiMessage,
DocumentAiRunRequest,
LocalBridgeRuntime,
build_agent_usage_from_response_usage,
build_document_ai_config_payload,
build_document_model_provider,
build_document_agent_instructions,
build_forward_headers,
run_document_agent_stream,
resolve_document_agent_request_model,
resolve_document_agent_model,
)
class AiDocumentAgentTests(unittest.TestCase):
def test_build_document_agent_instructions_includes_core_context(self):
request = DocumentAiRunRequest(
userId="user-1",
messages=[DocumentAiMessage(role="user", content="帮我改写第一段")],
context=DocumentAiContextPayload(
documentId="doc-1",
documentBlocks=[
{
"id": "block-1",
"type": "paragraph",
"content": [{"text": "第一段正文"}],
"children": [],
}
],
pageOptions={"wideLayout": True},
outline={"items": ["第一段"]},
),
)
instructions = build_document_agent_instructions(request)
self.assertIn("documentId=doc-1", instructions)
self.assertIn("documentBlockSummaries=", instructions)
self.assertIn("第一段正文", instructions)
self.assertIn("pageOptions=", instructions)
self.assertIn("outline=", instructions)
def test_build_document_agent_instructions_includes_profile_policy(self):
request = DocumentAiRunRequest(
userId="user-1",
profileId="page_writer_ai_first",
messages=[DocumentAiMessage(role="user", content="帮我补一段总结")],
context=DocumentAiContextPayload(
documentId="doc-1",
documentBlocks=[],
),
)
instructions = build_document_agent_instructions(request)
self.assertIn("AI 优先直接进入主编辑区起草", instructions)
self.assertIn("documentId=doc-1", instructions)
def test_build_document_agent_instructions_includes_title_write_hard_rules(self):
request = DocumentAiRunRequest(
userId="user-1",
messages=[DocumentAiMessage(role="user", content="请把当前页面标题改成“新标题”。")],
context=DocumentAiContextPayload(
documentId="doc-1",
documentBlocks=[],
),
)
instructions = build_document_agent_instructions(request)
self.assertIn("你必须调用一次 slash_run", instructions)
self.assertIn("/rename doc-1 <新标题>", instructions)
def test_build_forward_headers_only_keeps_allowed_headers(self):
forwarded = build_forward_headers(
{
"authorization": "Bearer token",
"cookie": "a=1",
"x-request-id": "req-1",
"x-trace-id": "trace-1",
"x-mnote-source-channel": "next_route",
"user-agent": "Vitest",
"x-ignore-me": "nope",
}
)
self.assertEqual(forwarded["authorization"], "Bearer token")
self.assertEqual(forwarded["cookie"], "a=1")
self.assertEqual(forwarded["x-request-id"], "req-1")
self.assertNotIn("x-ignore-me", forwarded)
def test_resolve_document_agent_model_uses_prefixed_request_model_first(self):
resolved = resolve_document_agent_model(
request_model="gh/gpt-4.1",
configured_default_model="ju/gpt-5.4",
env={},
available_models_loader=lambda _base_url, _api_key: [],
)
self.assertEqual(resolved, "gh/gpt-4.1")
def test_resolve_document_agent_request_model_prefers_model_key_semantic_name(self):
resolved = resolve_document_agent_request_model(
request_model=None,
request_model_key="gpt-5.3-codex",
configured_default_model="ju/gpt-5.4",
env={
"OPENAI_BASE_URL": "http://127.0.0.1:20128/v1",
"OPENAI_API_KEY": "sk-test",
},
available_models_loader=lambda _base_url, _api_key: [
"slow",
"fast",
"codex",
"ju/gpt-5.4",
],
)
self.assertEqual(resolved, "gpt-5.3-codex")
def test_resolve_document_agent_model_prefers_responses_provider_for_bare_default(self):
resolved = resolve_document_agent_model(
request_model=None,
configured_default_model="",
env={
"CLAW_DEFAULT_MODEL": "gpt-5.4",
"OPENAI_BASE_URL": "http://127.0.0.1:20128/v1",
"OPENAI_API_KEY": "sk-test",
},
available_models_loader=lambda _base_url, _api_key: [
"gh/gpt-5.4",
"ju/gpt-5.4",
"gmn/gpt-5.4",
],
)
self.assertEqual(resolved, "ju/gpt-5.4")
def test_resolve_document_agent_model_requires_explicit_prefix_when_bare_model_still_ambiguous(self):
with self.assertRaisesRegex(RuntimeError, "provider/model"):
resolve_document_agent_model(
request_model=None,
configured_default_model="",
env={
"CLAW_DEFAULT_MODEL": "gpt-5.4",
"OPENAI_BASE_URL": "http://127.0.0.1:20128/v1",
"OPENAI_API_KEY": "sk-test",
},
available_models_loader=lambda _base_url, _api_key: [
"gh/gpt-5.4",
"gmn/gpt-5.4",
],
)
def test_resolve_document_agent_model_fails_when_no_model_is_configured(self):
with self.assertRaisesRegex(RuntimeError, "未配置文档页 AI model"):
resolve_document_agent_model(
request_model=None,
configured_default_model="",
env={},
available_models_loader=lambda _base_url, _api_key: [],
)
def test_build_document_model_provider_supports_gateway_prefixed_models(self):
provider = build_document_model_provider(
request_model="ju/gpt-5.4",
configured_default_model="",
env={
"OPENAI_BASE_URL": "http://127.0.0.1:20128/v1",
"OPENAI_API_KEY": "sk-test",
},
available_models_loader=lambda _base_url, _api_key: [
"ju/gpt-5.4",
"gh/gpt-4.1",
],
)
ju_model = provider.get_model("ju/gpt-5.4")
gh_model = provider.get_model("gh/gpt-4.1")
self.assertEqual(getattr(ju_model, "model", None), "ju/gpt-5.4")
self.assertEqual(getattr(gh_model, "model", None), "gh/gpt-4.1")
def test_build_agent_usage_from_response_usage_falls_back_to_input_plus_output(self):
class FakeUsage:
input_tokens = 12
output_tokens = 5
total_tokens = None
input_tokens_details = None
output_tokens_details = None
usage = build_agent_usage_from_response_usage(FakeUsage())
self.assertEqual(usage.input_tokens, 12)
self.assertEqual(usage.output_tokens, 5)
self.assertEqual(usage.total_tokens, 17)
def test_build_agent_usage_from_response_usage_ignores_invalid_total_tokens(self):
class FakeUsage:
input_tokens = 8
output_tokens = 3
total_tokens = "not-a-number"
input_tokens_details = None
output_tokens_details = None
usage = build_agent_usage_from_response_usage(FakeUsage())
self.assertEqual(usage.input_tokens, 8)
self.assertEqual(usage.output_tokens, 3)
self.assertEqual(usage.total_tokens, 11)
def test_build_document_ai_config_payload_exposes_model_keys_profiles_and_tools(self):
payload = build_document_ai_config_payload(
env={
"OPENAI_BASE_URL": "http://127.0.0.1:20128/v1",
"CLAW_DEFAULT_MODEL": "gpt-5.4",
}
)
self.assertEqual(payload["provider"], "online")
self.assertEqual(payload["baseUrl"], "http://localhost:20128/v1")
self.assertEqual(payload["defaultModelKey"], "gpt-5.4")
self.assertEqual(payload["defaultProfileId"], "page_writer_ai_first")
self.assertTrue(payload["sessionEnabled"])
self.assertTrue(any(item["key"] == "gpt-5.4" for item in payload["models"]))
self.assertTrue(
any(
item["key"] == "gpt-5.4"
and item["resolvedCombo"] == "slow"
and item["resolvedRuntimeModel"] == "gpt-5.4"
for item in payload["models"]
)
)
self.assertTrue(any(item["id"] == "page_writer_ai_first" for item in payload["profiles"]))
self.assertTrue(any(item["name"] == "doc_get" for item in payload["tools"]))
class AiDocumentAgentBridgeRuntimeTests(unittest.IsolatedAsyncioTestCase):
async def test_local_bridge_runtime_execute_tool_uses_local_runtime_result_mode(self):
bridge = LocalBridgeRuntime()
ctx = DocumentAiRunContext(
user_id="user-1",
request_id="req-1",
trace_id="trace-1",
session_id=None,
workspace_id="ws-1",
document_id="doc-1",
document_blocks=[],
node=None,
subtree=None,
outline=None,
evidence=None,
page_options=None,
editor_runtime_page_options=None,
forward_headers={},
emitter=DocumentAiEmitter(),
bridge=bridge,
)
mock_run_bridge_runtime = AsyncMock(
return_value={
"ok": True,
"result": {
"ok": True,
"parsed": {
"command": "rename_doc",
},
},
}
)
with patch(
"app.services.ai_document_agent.run_bridge_runtime",
new=mock_run_bridge_runtime,
):
result = await bridge.execute_tool(
tool="slash_run",
invocation_kind="command",
context=ctx,
args_json={"text": "/rename doc-1 新标题"},
data={"source": "ai-orchestrator"},
target={"pageId": "doc-1"},
)
self.assertTrue(result["ok"])
self.assertEqual(result["parsed"]["command"], "rename_doc")
payload = mock_run_bridge_runtime.await_args.args[0]
self.assertEqual(payload["kind"], "tool")
self.assertEqual(payload["tool"]["tool"], "slash_run")
self.assertEqual(payload["tool"]["kind"], "command")
self.assertEqual(payload["tool"]["mode"], "result")
self.assertEqual(payload["tool"]["target"]["pageId"], "doc-1")
self.assertEqual(payload["context"]["source"]["channel"], "ai_orchestrator")
self.assertEqual(payload["context"]["source"]["client"], "wolai-backend")
async def test_run_document_agent_stream_does_not_require_legacy_mnote_web_config(self):
request = DocumentAiRunRequest(
userId="user-1",
maxSteps=1,
messages=[DocumentAiMessage(role="user", content="请总结当前页面")],
context=DocumentAiContextPayload(
documentId="doc-1",
documentBlocks=[],
),
)
class FakeRunner:
@staticmethod
async def run(*_args, **_kwargs):
return SimpleNamespace(final_output="已完成")
with (
patch("app.services.ai_document_agent.sdk_available", return_value=True),
patch(
"app.services.ai_document_agent.has_configured_openai_api_key",
return_value=True,
),
patch(
"app.services.ai_document_agent.build_document_agent",
return_value=object(),
),
patch(
"app.services.ai_document_agent.build_document_model_provider",
return_value=object(),
),
patch(
"app.services.ai_document_agent.build_session",
return_value=None,
),
patch(
"app.services.ai_document_agent.RunConfig",
side_effect=lambda **kwargs: kwargs,
),
patch(
"app.services.ai_document_agent.Runner",
new=FakeRunner,
),
patch(
"app.services.ai_document_agent.trace",
new=lambda *_args, **_kwargs: contextlib.nullcontext(),
),
):
chunks: list[str] = []
async for chunk in run_document_agent_stream(request, source_headers={}):
chunks.append(chunk)
text = "".join(chunks)
self.assertIn("event: assistant_message", text)
self.assertIn("event: completion", text)
self.assertIn("已完成", text)
if __name__ == "__main__":
unittest.main()
-4
View File
@@ -1,4 +0,0 @@
INFO: Started server process [27148]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
View File
-4
View File
@@ -1,4 +0,0 @@
INFO: Started server process [10528]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
-30
View File
@@ -1,30 +0,0 @@
INFO: 127.0.0.1:56337 - "GET / HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 183.221.18.13:0 - "GET /health HTTP/1.1" 200 OK
INFO: 183.221.18.13:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 183.221.18.13:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
INFO: 2409:8a62:e45:7c51::16f3:0 - "GET /health HTTP/1.1" 200 OK
View File
View File
-57
View File
@@ -1,57 +0,0 @@
INFO: Will watch for changes in these directories: ['F:\\SOFT\\MNOTE\\wolai-backend']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [20384] using WatchFiles
INFO: Started server process [6920]
INFO: Waiting for application startup.
INFO: Application startup complete.
WARNING: WatchFiles detected changes in 'app\main.py'. Reloading...
INFO: Shutting down
INFO: Waiting for application shutdown.
INFO: Application shutdown complete.
INFO: Finished server process [6920]
INFO: Started server process [5608]
INFO: Waiting for application startup.
INFO: Application startup complete.
WARNING: WatchFiles detected changes in 'app\deps.py'. Reloading...
INFO: Shutting down
INFO: Waiting for application shutdown.
INFO: Application shutdown complete.
INFO: Finished server process [5608]
WARNING: WatchFiles detected changes in 'app\deps.py', 'app\routers\tasks.py', 'app\workers\tasks.py', 'app\services\task_tracker.py'. Reloading...
INFO: Started server process [33148]
INFO: Waiting for application startup.
INFO: Application startup complete.
ERROR: Traceback (most recent call last):
File "C:\Users\liaib\AppData\Local\Programs\Python\Python311\Lib\asyncio\runners.py", line 118, in run
return self._loop.run_until_complete(task)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\liaib\AppData\Local\Programs\Python\Python311\Lib\asyncio\base_events.py", line 654, in run_until_complete
return future.result()
^^^^^^^^^^^^^^^
asyncio.exceptions.CancelledError
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "C:\Users\liaib\AppData\Local\Programs\Python\Python311\Lib\asyncio\runners.py", line 190, in run
return runner.run(main)
^^^^^^^^^^^^^^^^
File "C:\Users\liaib\AppData\Local\Programs\Python\Python311\Lib\asyncio\runners.py", line 123, in run
raise KeyboardInterrupt()
KeyboardInterrupt
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "F:\SOFT\MNOTE\wolai-backend\venv\Lib\site-packages\starlette\routing.py", line 700, in lifespan
await receive()
File "F:\SOFT\MNOTE\wolai-backend\venv\Lib\site-packages\uvicorn\lifespan\on.py", line 137, in receive
return await self.receive_queue.get()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\liaib\AppData\Local\Programs\Python\Python311\Lib\asyncio\queues.py", line 158, in get
await getter
asyncio.exceptions.CancelledError
INFO: Started server process [12156]
INFO: Waiting for application startup.
INFO: Application startup complete.
-1
View File
@@ -1 +0,0 @@
INFO: 127.0.0.1:57108 - "GET /health HTTP/1.1" 200 OK