diff --git a/app/actions/update.py b/app/actions/update.py index a445625..50b2171 100644 --- a/app/actions/update.py +++ b/app/actions/update.py @@ -3,6 +3,7 @@ from copy import deepcopy from app.actions.base import BaseActionHandler +from app.resources.material_fields import editable_fields, known_prop_keys from app.schemas.actions import UpdateAction from app.schemas.screen import Screen @@ -16,6 +17,19 @@ class UpdateHandler(BaseActionHandler[UpdateAction]): if action.section == "style" and target.style is None: target.style = {} + # props 第一层不能新造字段:渲染器不会读取约定外的字段,新造只会“成功”却没有效果 + if action.section == "props": + known = known_prop_keys(target) + if known is not None and action.path[0] not in known: + labels = "、".join( + f"{field['label']}={field['section']}.{field['path']}" + for field in editable_fields(target.type) + ) + raise ValueError( + f"组件 {target.type} 的 props 没有字段 {action.path[0]!r}," + f"不能新造组件级属性;可编辑字段:{labels or '无'}" + ) + # 获取动作指定的属性区域,作为路径遍历的起点 current = getattr(target, action.section) # 逐段访问路径,只有到达末级时才写入新值 diff --git a/app/api/agent.py b/app/api/agent.py index acbdc28..6e0edfd 100644 --- a/app/api/agent.py +++ b/app/api/agent.py @@ -1,24 +1,34 @@ """大屏 Agent 启动与澄清恢复接口""" -from fastapi import APIRouter, Request +from fastapi import APIRouter, Depends, Request from app.common.sse import AgentEventResponse +from app.core.auth import get_current_user from app.schemas.agent import ResumeRequest from app.schemas.request import DesignRequest +from app.schemas.user import CurrentUser router = APIRouter(prefix="/agent", tags=["Agent"]) @router.post("/run", response_class=AgentEventResponse, status_code=200) -async def stream_agent(body: DesignRequest, request: Request): +async def stream_agent( + body: DesignRequest, + request: Request, + user: CurrentUser = Depends(get_current_user), +): service = request.app.state.agent_service # 在发送流响应头之前完成会话校验并占用任务 - execution = await service.prepare_run(body) + execution = await service.prepare_run(body, user.id) return AgentEventResponse(service, execution, request) @router.post("/resume", response_class=AgentEventResponse, status_code=200) -async def stream_resume_agent(body: ResumeRequest, request: Request): +async def stream_resume_agent( + body: ResumeRequest, + request: Request, + user: CurrentUser = Depends(get_current_user), +): service = request.app.state.agent_service - execution = await service.prepare_resume(body) + execution = await service.prepare_resume(body, user.id) return AgentEventResponse(service, execution, request) diff --git a/app/api/conversations.py b/app/api/conversations.py index e55328e..013787a 100644 --- a/app/api/conversations.py +++ b/app/api/conversations.py @@ -1,28 +1,35 @@ """会话创建与历史查询接口""" -from fastapi import APIRouter, Request +from fastapi import APIRouter, Depends, Request from app.common.response import ApiResponse, success_response +from app.core.auth import get_current_user from app.core.exceptions import BizException from app.schemas.conversation import ConversationCreated, ConversationDetail +from app.schemas.user import CurrentUser router = APIRouter(prefix="/conversations", tags=["Conversation"]) @router.post("") async def create_conversation( - request: Request, + request: Request, user: CurrentUser = Depends(get_current_user) ) -> ApiResponse[ConversationCreated]: store = request.app.state.conversation_store - return success_response(ConversationCreated(conversation_id=await store.create())) + return success_response( + ConversationCreated(conversation_id=await store.create(user.id)) + ) @router.get("/{conversation_id}") async def get_conversation( - conversation_id: str, request: Request + conversation_id: str, + request: Request, + user: CurrentUser = Depends(get_current_user), ) -> ApiResponse[ConversationDetail]: store = request.app.state.conversation_store - if not await store.exists(conversation_id): + # 他人的会话与不存在同样返回 404 + if not await store.exists(conversation_id, user.id): raise BizException(404, "会话不存在或已过期") turns = await store.load_turns(conversation_id, None) return success_response( diff --git a/app/clients/user_center.py b/app/clients/user_center.py new file mode 100644 index 0000000..5068b5c --- /dev/null +++ b/app/clients/user_center.py @@ -0,0 +1,117 @@ +"""用户中心客户端:用请求携带的 token 换取当前用户,结果短时缓存""" + +import hashlib +import time +from functools import cache + +import httpx +from loguru import logger + +from app.config import Settings, settings +from app.core.exceptions import BizException +from app.schemas.user import CurrentUser + +# 用户中心 401 时提示调用方用 Bearer token 重新认证 +BEARER_CHALLENGE = {"WWW-Authenticate": "Bearer"} + + +class UserCenterClient: + def __init__( + self, + config: Settings | None = None, + transport: httpx.AsyncBaseTransport | None = None, + ): + self._config = config if config is not None else settings + self._transport = transport + self._client: httpx.AsyncClient | None = None + # token 摘要 -> (过期时刻, 用户);不在内存中保存明文 token + self._cache: dict[str, tuple[float, CurrentUser]] = {} + + def _get_client(self) -> httpx.AsyncClient: + if self._client is None: + self._client = httpx.AsyncClient( + base_url=self._config.user_center_url, + timeout=self._config.user_center_timeout_seconds, + transport=self._transport, + ) + return self._client + + def _cached(self, key: str) -> CurrentUser | None: + entry = self._cache.get(key) + if entry is None: + return None + if entry[0] <= time.monotonic(): + del self._cache[key] + return None + return entry[1] + + def _remember(self, key: str, user: CurrentUser) -> None: + ttl = self._config.auth_cache_ttl_seconds + if not ttl: + return + now = time.monotonic() + # 写入时顺带清掉过期项,缓存规模不超过 TTL 内的活跃 token 数 + for stale in [k for k, (expires, _) in self._cache.items() if expires <= now]: + del self._cache[stale] + self._cache[key] = (now + ttl, user) + + async def fetch_current_user(self, token: str) -> CurrentUser: + key = hashlib.sha256(token.encode()).hexdigest() + user = self._cached(key) + if user is not None: + return user + + try: + response = await self._get_client().get( + "/api/users/me", headers={"Authorization": f"Bearer {token}"} + ) + except httpx.HTTPError as error: + logger.opt(exception=error).warning("用户中心请求失败") + raise BizException(503, "用户中心不可用") from error + + if response.status_code >= 500: + logger.warning("用户中心返回 HTTP {}", response.status_code) + raise BizException(503, "用户中心不可用") + try: + body = response.json() + except ValueError: + body = None + if not isinstance(body, dict): + body = {} + if response.status_code >= 400: + # 用户中心与本服务的错误响应同构,直接透传状态码和提示 + raise BizException( + response.status_code, + body.get("msg") or "认证失败", + BEARER_CHALLENGE if response.status_code == 401 else None, + ) + try: + user = CurrentUser.model_validate(body["data"]) + except (KeyError, TypeError, ValueError) as error: + logger.warning("用户中心返回的数据格式无效") + raise BizException(503, "用户中心不可用") from error + + self._remember(key, user) + return user + + async def aclose(self) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + + +@cache +def _cached_client() -> UserCenterClient: + return UserCenterClient() + + +def get_user_center() -> UserCenterClient: + """共享入口;适用于单个服务事件循环""" + return _cached_client() + + +async def close_user_center() -> None: + """在应用关闭阶段调用""" + if _cached_client.cache_info().currsize: + await _cached_client().aclose() + _cached_client.cache_clear() diff --git a/app/config.py b/app/config.py index 0446474..40d6249 100644 --- a/app/config.py +++ b/app/config.py @@ -79,7 +79,32 @@ class Settings: default_factory=lambda: int(os.getenv("MONGODB_CHECKPOINT_TTL", "604800")) ) + # 用户中心:请求携带的 token 交给它校验,换回 user_id + user_center_url: str = field( + default_factory=lambda: ( + os.getenv("USER_CENTER_URL") or "http://127.0.0.1:8100" + ).rstrip("/") + ) + user_center_timeout_seconds: float = field( + default_factory=lambda: float(os.getenv("USER_CENTER_TIMEOUT_SECONDS", "5")) + ) + # 关闭后不校验 token,所有请求视为同一个本地调试用户,仅用于开发 + auth_enabled: bool = field( + default_factory=lambda: ( + os.getenv("AUTH_ENABLED", "true").strip().lower() + not in ("false", "0", "no", "off") + ) + ) + # token 校验结果缓存时长(秒),登出或禁用后最多延迟这么久失效,0 表示不缓存 + auth_cache_ttl_seconds: int = field( + default_factory=lambda: int(os.getenv("AUTH_CACHE_TTL_SECONDS", "30")) + ) + def __post_init__(self): + if self.user_center_timeout_seconds <= 0: + raise ValueError("USER_CENTER_TIMEOUT_SECONDS 必须是正数") + if self.auth_cache_ttl_seconds < 0: + raise ValueError("AUTH_CACHE_TTL_SECONDS 必须是非负整数") if self.mongodb_checkpoint_ttl < 0: raise ValueError("MONGODB_CHECKPOINT_TTL 必须是非负整数") if self.conversation_empty_ttl <= 0: diff --git a/app/core/auth.py b/app/core/auth.py new file mode 100644 index 0000000..df963bf --- /dev/null +++ b/app/core/auth.py @@ -0,0 +1,28 @@ +"""接口鉴权:从 Authorization 头取出用户中心 token,换取当前用户""" + +from fastapi import Depends, Request + +from app.clients.user_center import BEARER_CHALLENGE, UserCenterClient, get_user_center +from app.config import settings +from app.core.exceptions import BizException +from app.schemas.user import CurrentUser + +# AUTH_ENABLED=false 时所有请求共用的本地调试用户 +DEV_USER = CurrentUser(id=0, username="dev") + + +def _bearer_token(request: Request) -> str: + scheme, _, token = request.headers.get("Authorization", "").partition(" ") + token = token.strip() + if scheme.lower() != "bearer" or not token: + raise BizException(401, "请先登录", BEARER_CHALLENGE) + return token + + +async def get_current_user( + request: Request, client: UserCenterClient = Depends(get_user_center) +) -> CurrentUser: + """同一请求内 FastAPI 会缓存结果,路由级和函数级重复声明只校验一次""" + if not settings.auth_enabled: + return DEV_USER + return await client.fetch_current_user(_bearer_token(request)) diff --git a/app/core/exceptions.py b/app/core/exceptions.py index 46940c2..b16d68a 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -16,14 +16,25 @@ from app.exceptions import AppError, DecisionError class BizException(AppError): """可安全返回给调用方的业务错误""" - def __init__(self, code: int, msg: str = "请求失败") -> None: + def __init__( + self, + code: int, + msg: str = "请求失败", + headers: dict[str, str] | None = None, + ) -> None: self.code = code self.msg = msg + # 协议响应头,例如 401 时的 WWW-Authenticate + self.headers = headers super().__init__(msg) async def handle_biz_exception(_request: Request, exc: BizException) -> JSONResponse: - return JSONResponse(status_code=exc.code, content=error_content(exc.code, exc.msg)) + return JSONResponse( + status_code=exc.code, + content=error_content(exc.code, exc.msg), + headers=exc.headers, + ) async def handle_validation_exception( diff --git a/app/main.py b/app/main.py index 7b6c98f..6c53fb9 100644 --- a/app/main.py +++ b/app/main.py @@ -3,12 +3,13 @@ import asyncio from contextlib import asynccontextmanager -from fastapi import FastAPI +from fastapi import Depends, FastAPI from loguru import logger from app.api.agent import router from app.api.conversations import router as conversation_router from app.api.materials import router as material_router +from app.core.auth import get_current_user from app.core.exceptions import register_exception_handlers from app.core.logging import setup_logging @@ -17,6 +18,7 @@ from app.core.logging import setup_logging async def lifespan(application: FastAPI): setup_logging() from app.clients.decision import close_decision_engine + from app.clients.user_center import close_user_center from app.observability.tracing import flush_tracing from app.processor.graph import build_graph from app.services.agent import AgentService @@ -42,9 +44,12 @@ async def lifespan(application: FastAPI): await close_decision_engine() finally: try: - await asyncio.to_thread(flush_tracing) + await close_user_center() finally: - await logger.complete() + try: + await asyncio.to_thread(flush_tracing) + finally: + await logger.complete() def create_app() -> FastAPI: @@ -52,8 +57,12 @@ def create_app() -> FastAPI: # 统一处理请求校验、HTTP 错误和未捕获异常 register_exception_handlers(application) # 统一 /api 前缀,方便前端代理和反向代理只配置一条规则 - application.include_router(router, prefix="/api") - application.include_router(conversation_router, prefix="/api") + # 会话和 agent 接口需登录,物料目录是全局公开数据 + authenticated = [Depends(get_current_user)] + application.include_router(router, prefix="/api", dependencies=authenticated) + application.include_router( + conversation_router, prefix="/api", dependencies=authenticated + ) application.include_router(material_router, prefix="/api") return application diff --git a/app/observability/tracing.py b/app/observability/tracing.py index c907801..581bce9 100644 --- a/app/observability/tracing.py +++ b/app/observability/tracing.py @@ -61,24 +61,28 @@ async def ainvoke_screen( *, thread_id: str, session_id: str | None = None, + user_id: int | None = None, ) -> Any: """每轮调用记录一个根 span,返回、等待用户或异常时结束。 session_id 由调用方在首次与恢复时一致传入(通常为 conversation_id)。 未传时始终使用 thread_id,不从仅首次请求才有的 payload 推断。 + user_id 为用户中心的用户 ID,接口调用时传入,本地脚本不传。 """ - with _screen_trace(payload, thread_id, session_id) as (config, root): + with _screen_trace(payload, thread_id, session_id, user_id) as (config, root): result = await graph.ainvoke(payload, config=config) _finish_trace(root, result) return result -async def astream_screen(graph, payload, *, thread_id, session_id=None): +async def astream_screen( + graph, payload, *, thread_id, session_id=None, user_id=None +): """同一轮流保持同一个追踪上下文,关闭流时结束追踪""" from contextlib import aclosing result = {} - with _screen_trace(payload, thread_id, session_id) as (config, root): + with _screen_trace(payload, thread_id, session_id, user_id) as (config, root): async with aclosing( graph.astream(payload, config=config, stream_mode=["custom", "values"]) ) as events: @@ -98,7 +102,7 @@ def _finish_trace(root, result): @contextmanager -def _screen_trace(payload, thread_id, session_id): +def _screen_trace(payload, thread_id, session_id, user_id=None): config = {"configurable": {"thread_id": thread_id}} client = _ensure_client() if tracing_enabled() else None if client is None: @@ -128,6 +132,8 @@ def _screen_trace(payload, thread_id, session_id): propagate_attributes( trace_name=TRACE_NAME, session_id=session_id or thread_id, + # 用户中心的 user_id,面板 Users 页按用户聚合;脚本调用没有用户时不设 + user_id=str(user_id) if user_id is not None else None, tags=[FEATURE_TAG], metadata={ "thread_id": thread_id, diff --git a/app/processor/nodes/action_node.py b/app/processor/nodes/action_node.py index fe7e912..d3a6bc1 100644 --- a/app/processor/nodes/action_node.py +++ b/app/processor/nodes/action_node.py @@ -11,6 +11,7 @@ from app.clients.llm import get_llm from app.processor.base import BaseNode from app.processor.state import QueryState from app.prompt.action import build_action_prompt +from app.resources.material_fields import editable_fields from app.schemas.action_validation import ActionValidationError, schema_errors from app.schemas.actions import ActionBatch, AddAction from app.schemas.operations import is_direct_operation @@ -58,6 +59,28 @@ def action_block_reason(state: QueryState) -> str | None: return None +def update_fields(state: QueryState) -> dict[str, list[dict]]: + """按组件 ID 汇总可能被属性更新的已有组件的可编辑字段,没有字段的组件省略""" + plan = state.get("screen_plan") + if plan is None: + target_ids = [ + target + for change in state["requirement"].changes + if change.target_scope == "nodes" + for target in change.target_ids + ] + else: + target_ids = [ + item.target_id for item in plan.components if item.disposition == "modify" + ] + nodes = {node.id: node for node in state["input_context"].screen.nodes} + result = {} + for target in dict.fromkeys(target_ids): + if target in nodes and (fields := editable_fields(nodes[target].type)): + result[target] = fields + return result + + class ActionNode(BaseNode): name = "action_node" @@ -92,6 +115,10 @@ class ActionNode(BaseNode): payload["screen_plan"] = state["screen_plan"].model_dump(mode="json") payload["component_materials"] = plan_materials(state) + # 提供待修改组件的属性面板字段,让模型把用户说的“标题”“颜色”等对应到真实路径 + if fields := update_fields(state): + payload["editable_fields"] = fields + # 如果本轮是动作修正,提供修正建议、上一轮动作和校验错误供模型参考 if state.get("repair_plan") is not None: payload.update( diff --git a/app/prompt/action_rules.py b/app/prompt/action_rules.py index b60436b..96fcfce 100644 --- a/app/prompt/action_rules.py +++ b/app/prompt/action_rules.py @@ -24,6 +24,8 @@ properties 只包含需要修改的画布顶层字段,宽高为最终绝对值 依据快照选择 props / style 下的字段路径,不猜测不存在的嵌套结构 path 中字符串是对象键,整数是数组下标;中间路径必须存在,数组下标不能越界 只替换指定字段,不做深度合并;null 表示赋值,不表示删除 +提供 editable_fields 时,它是该组件属性面板的字段:用户说的名称(如名字、标题、颜色)优先对应 label 相符的字段,使用其 section 和 path +props 第一层只能使用组件已有或物料默认配置中的字段,不新造组件级属性;更深层的配置可以按需新增 不通过修改属性改变组件类型(如把 series 的 type 改成其他图表类型),组件类型由节点 type 和物料决定 style 为 null 时可用单层路径添加属性,保留未要求修改的配置 """, diff --git a/app/resources/component.py b/app/resources/component.py index 49f0d44..f87878c 100644 --- a/app/resources/component.py +++ b/app/resources/component.py @@ -1,4 +1,4 @@ -"""读取前端组件目录:启动时载入进程内存缓存,未载入时回退读取本地 JSON 文件""" +"""查询前端组件目录:目录只来自启动时从 MongoDB 载入的进程内存缓存,没有文件回退""" from pathlib import Path @@ -8,8 +8,6 @@ from app.schemas.component_resource import ( ComponentResult, ) -DEFAULT_CATALOG_PATH = Path(__file__).resolve().parent / "catalog" / "components.json" - # 进程内共享的组件目录,由启动流程载入;agent 与前端接口读取同一份 _cached_catalog: ComponentCatalog | None = None @@ -23,36 +21,24 @@ def get_catalog_cache() -> ComponentCatalog | None: return _cached_catalog -def load_catalog_file(path: str | Path | None = None) -> ComponentCatalog | None: +def load_catalog_file(path: str | Path) -> ComponentCatalog: + """读取导出的目录文件,只供导入脚本使用,运行时不读取文件""" # 如果传入的是相对路径,以项目根目录为基准,避免受启动时工作目录影响 - target = Path(path) if path is not None else DEFAULT_CATALOG_PATH + target = Path(path) if not target.is_absolute(): target = Path(__file__).resolve().parents[2] / target - try: - # utf-8-sig 同时兼容带 BOM 和不带 BOM 的 UTF-8 文件 - content = target.read_text(encoding="utf-8-sig") - except FileNotFoundError: - return None - # 文件损坏或定义不合法直接报错,不能伪装成查无结果 - return ComponentCatalog.model_validate_json(content) + # utf-8-sig 同时兼容带 BOM 和不带 BOM 的 UTF-8 文件;损坏或不合法直接报错 + return ComponentCatalog.model_validate_json(target.read_text(encoding="utf-8-sig")) class ComponentProvider: - def __init__(self, catalog_path: str | Path | None = None): - # 指定路径时始终读取该文件;未指定时优先使用启动缓存,缓存为空再读默认文件 - self.catalog_path = catalog_path - def query(self, query: ComponentQuery) -> ComponentResult: - catalog = ( - _cached_catalog - if self.catalog_path is None and _cached_catalog is not None - else load_catalog_file(self.catalog_path) - ) + catalog = _cached_catalog if catalog is None: - # 如果目录文件不存在,返回未支持状态和原因,由调用方决定失败处理方式 + # 启动时未载入目录,返回未支持状态和原因,由调用方决定失败处理方式 return ComponentResult( status="unsupported", - message="尚未配置前端组件目录,请先导出真实组件定义", + message="物料目录未载入,请确认已配置 MongoDB 并导入物料", ) # 如果提供关键词,去除首尾空白并统一大小写,未提供时不启用关键词筛选 @@ -92,9 +78,3 @@ class ComponentProvider: catalog_version=catalog.version, message="已找到组件定义" if matches else "已查询组件目录,未找到匹配的组件", ) - - -if __name__ == "__main__": - provider = ComponentProvider() - result = provider.query(ComponentQuery()) - print(result.model_dump_json(indent=2)) diff --git a/app/resources/material_fields.py b/app/resources/material_fields.py new file mode 100644 index 0000000..be11e34 --- /dev/null +++ b/app/resources/material_fields.py @@ -0,0 +1,50 @@ +"""从物料目录读取组件的可编辑字段,供动作生成理解字段含义,并供校验约束属性结构""" + +from app.resources.component import ComponentProvider +from app.schemas.component_resource import ComponentDefinition, ComponentQuery +from app.schemas.screen import ScreenNode + +# 属性面板的 key 以区域开头,只有这两个区域可以被 update 修改 +EDITABLE_SECTIONS = ("props", "style") + + +def _materials(node_type: str) -> list[ComponentDefinition]: + # 多个物料预设可以共用同一前端 type,按 type 取全部预设 + result = ComponentProvider().query(ComponentQuery(type=node_type)) + return result.components if result.status == "resolved" else [] + + +def _split_key(key: str) -> tuple[str, list[str | int]] | None: + # 前端点路径中的纯数字段是数组下标,转换为与 UpdateAction.path 一致的整数 + section, *path = key.split(".") + if section not in EDITABLE_SECTIONS or not path: + return None + return section, [int(part) if part.isdigit() else part for part in path] + + +def editable_fields(node_type: str) -> list[dict]: + """返回组件属性面板中的字段:用户可见名称、区域、路径及可选值,按出现顺序去重""" + fields: dict[str, dict] = {} + for material in _materials(node_type): + for setter in material.setters: + parsed = _split_key(setter.key) + if parsed is None or setter.key in fields: + continue + section, path = parsed + field = {"label": setter.label, "section": section, "path": path} + if options := setter.props.get("options"): + field["options"] = options + fields[setter.key] = field + return list(fields.values()) + + +def known_prop_keys(node: ScreenNode) -> set[str] | None: + """props 第一层是组件与渲染器的约定,只能是节点已有或物料默认配置中的字段; + 目录中找不到该类型时约定未知,返回 None 表示不做限制""" + materials = _materials(node.type) + if not materials: + return None + keys = set(node.props) + for material in materials: + keys.update(material.schema_.props) + return keys diff --git a/app/schemas/user.py b/app/schemas/user.py new file mode 100644 index 0000000..277d590 --- /dev/null +++ b/app/schemas/user.py @@ -0,0 +1,8 @@ +"""用户中心返回的当前用户,只保留本服务用到的字段""" + +from pydantic import BaseModel + + +class CurrentUser(BaseModel): + id: int + username: str diff --git a/app/services/agent.py b/app/services/agent.py index 3bc14be..c97bfa4 100644 --- a/app/services/agent.py +++ b/app/services/agent.py @@ -31,12 +31,12 @@ class AgentService: self._active: set[str] = set() self._conversation_of: dict[str, str] = {} - async def prepare_run(self, request: DesignRequest): + async def prepare_run(self, request: DesignRequest, user_id: int): # 会话 ID 必须由后端签发,不再为缺失的 ID 兜底生成 conversation_id = request.context.conversation_id if conversation_id is None: raise BizException(422, "请先创建会话并传入 conversation_id") - if not await self.store.exists(conversation_id): + if not await self.store.exists(conversation_id, user_id): raise BizException(404, "会话不存在或已过期") # 新请求始终独立建任务,不用 conversation_id 充当检查点 ID @@ -72,9 +72,10 @@ class AgentService: "message_id": str(uuid4()), }, conversation_id, + user_id, ) - async def prepare_resume(self, request: ResumeRequest): + async def prepare_resume(self, request: ResumeRequest, user_id: int): # 在发送流响应头之前完成校验并占用任务 thread_id = str(request.thread_id) # 检查和占用之间没有 await,同一事件循环内不会重复进入 @@ -83,7 +84,10 @@ class AgentService: snapshot = await self.graph.aget_state( {"configurable": {"thread_id": thread_id}} ) - if not snapshot.values: + # 不属于当前用户的任务与不存在同样处理,先于其余校验,不泄露任务状态 + if not snapshot.values or not await self.store.exists( + snapshot.values["conversation_id"], user_id + ): raise BizException(404, "任务不存在或已过期") if not any(task.interrupts for task in snapshot.tasks): raise BizException(409, "任务当前未等待用户补充") @@ -119,6 +123,7 @@ class AgentService: update={"message_id": str(uuid4())}, ), conversation_id, + user_id, ) except BaseException: self.release(thread_id) @@ -196,13 +201,17 @@ class AgentService: failure_reason=state.get("failure_reason"), ) - async def stream(self, thread_id, payload, session_id): + async def stream(self, thread_id, payload, session_id, user_id=None): """消费图事件,完整状态只用于整理最终结果,不直接暴露给客户端""" state = None with logger.contextualize(thread_id=thread_id): async with aclosing( astream_screen( - self.graph, payload, thread_id=thread_id, session_id=session_id + self.graph, + payload, + thread_id=thread_id, + session_id=session_id, + user_id=user_id, ) ) as events: async for mode, data in events: diff --git a/app/services/conversation.py b/app/services/conversation.py index 3b04425..9c57ce2 100644 --- a/app/services/conversation.py +++ b/app/services/conversation.py @@ -11,9 +11,11 @@ from app.schemas.conversation import Turn class ConversationStore(Protocol): """内存和 MongoDB 实现同一组方法,业务代码不需要关心具体存储""" - async def create(self) -> str: ... + async def create(self, user_id: int) -> str: ... - async def exists(self, conversation_id: str) -> bool: ... + async def exists(self, conversation_id: str, user_id: int) -> bool: + """会话已签发、未过期且属于该用户;不属于时与不存在同样返回 False""" + ... async def load_turns( self, conversation_id: str, limit: int | None @@ -32,6 +34,7 @@ class InMemoryConversationStore: self._empty_ttl = timedelta(seconds=ttl) # 会话 ID -> 创建时间;轮次按 thread_id 去重并保持首次写入顺序 self._created: dict[str, datetime] = {} + self._owner: dict[str, int] = {} self._turns: dict[str, dict[str, Turn]] = {} def _purge_expired(self) -> None: @@ -43,17 +46,22 @@ class InMemoryConversationStore: if created < deadline and not self._turns.get(key) ]: del self._created[conversation_id] + self._owner.pop(conversation_id, None) self._turns.pop(conversation_id, None) - async def create(self) -> str: + async def create(self, user_id: int) -> str: self._purge_expired() conversation_id = str(uuid4()) self._created[conversation_id] = _now() + self._owner[conversation_id] = user_id return conversation_id - async def exists(self, conversation_id: str) -> bool: + async def exists(self, conversation_id: str, user_id: int) -> bool: self._purge_expired() - return conversation_id in self._created + return ( + conversation_id in self._created + and self._owner.get(conversation_id) == user_id + ) async def load_turns(self, conversation_id: str, limit: int | None) -> list[Turn]: turns = list(self._turns.get(conversation_id, {}).values()) diff --git a/app/services/conversation_mongo.py b/app/services/conversation_mongo.py index 2bf54f6..35c565a 100644 --- a/app/services/conversation_mongo.py +++ b/app/services/conversation_mongo.py @@ -42,6 +42,8 @@ class MongoConversationStore: ) # 空会话清理按创建时间扫描 await self._conversations.create_index("created_at") + # 按用户查询会话 + await self._conversations.create_index("user_id") async def _purge_expired(self) -> None: # 只清理没有任何轮次且超过保留时长的空会话,有对话的会话不受影响 @@ -63,17 +65,20 @@ class MongoConversationStore: if empty: await self._conversations.delete_many({"_id": {"$in": empty}}) - async def create(self) -> str: + async def create(self, user_id: int) -> str: await self._purge_expired() conversation_id = str(uuid4()) - # user_id 为后续长期记忆预留,当前恒为空 + # user_id 来自用户中心,会话只对创建者可见 await self._conversations.insert_one( - {"_id": conversation_id, "created_at": _now(), "user_id": None} + {"_id": conversation_id, "created_at": _now(), "user_id": user_id} ) return conversation_id - async def exists(self, conversation_id: str) -> bool: - document = await self._conversations.find_one({"_id": conversation_id}) + async def exists(self, conversation_id: str, user_id: int) -> bool: + # 不属于该用户的会话与不存在同样处理,不泄露会话是否存在 + document = await self._conversations.find_one( + {"_id": conversation_id, "user_id": user_id} + ) if document is None: return False if document["created_at"] >= _now() - self._empty_ttl: diff --git a/app/services/material_mongo.py b/app/services/material_mongo.py index 7e9c5a4..b5844a8 100644 --- a/app/services/material_mongo.py +++ b/app/services/material_mongo.py @@ -3,7 +3,7 @@ from loguru import logger from pymongo.asynchronous.database import AsyncDatabase -from app.resources.component import load_catalog_file, set_catalog_cache +from app.resources.component import set_catalog_cache from app.schemas.component_resource import ComponentCatalog # 元数据记录的 _id,组件记录的 _id 为组件 key @@ -49,11 +49,13 @@ class MongoMaterialRepository: async def warm_material_cache(db: AsyncDatabase | None) -> ComponentCatalog | None: - """启动时把物料目录载入进程缓存;Mongo 未启用或尚未导入时回退读本地 JSON""" + """启动时把物料目录从 MongoDB 载入进程缓存;未启用或尚未导入时目录为空,不回退读文件""" catalog = await MongoMaterialRepository(db).load_all() if db is not None else None if catalog is None: - catalog = load_catalog_file() - logger.warning("物料未从 MongoDB 加载,回退使用本地目录文件") + logger.error( + "物料目录未载入:{},新增、更换组件等需要物料的修改将无法执行", + "未配置 MONGODB_URI" if db is None else "materials 集合尚未导入", + ) else: logger.info("已从 MongoDB 加载物料 {} 个", len(catalog.components)) set_catalog_cache(catalog) diff --git a/pyproject.toml b/pyproject.toml index 2d50bab..14d7226 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -15,4 +15,5 @@ dependencies = [ "loguru>=0.7.3", "pymongo>=4.17.0", "langgraph-checkpoint-mongodb>=0.5.0", + "httpx>=0.28.1", ] diff --git a/uv.lock b/uv.lock index b30ab57..5cbe572 100644 --- a/uv.lock +++ b/uv.lock @@ -1117,6 +1117,7 @@ version = "0.1.0" source = { virtual = "." } dependencies = [ { name = "fastapi", extra = ["standard"] }, + { name = "httpx" }, { name = "langchain" }, { name = "langchain-openai" }, { name = "langfuse" }, @@ -1133,6 +1134,7 @@ dependencies = [ [package.metadata] requires-dist = [ { name = "fastapi", extras = ["standard"], specifier = ">=0.141.1" }, + { name = "httpx", specifier = ">=0.28.1" }, { name = "langchain", specifier = ">=1.4.2" }, { name = "langchain-openai", specifier = ">=1.6.2" }, { name = "langfuse", specifier = "==4.15.6" },