| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758 |
- """按服务端会话身份映射文件根，虚拟路径保持不变。"""
- import hashlib
- from pathlib import Path
- from typing import Any
- import config
- def session_root(user_name: str, session_id: str) -> Path:
- # 哈希避免用户名或外部 ID 成为路径片段；不迁移旧共享文件以免猜错归属。
- user_key = hashlib.sha256(user_name.encode()).hexdigest()
- session_key = hashlib.sha256(session_id.encode()).hexdigest()
- root = config.ROOT / "data" / "workspaces" / user_key / session_key
- root.mkdir(parents=True, exist_ok=True)
- return root
- def _session_identity(runtime) -> tuple[str, str]:
- """取服务端会话身份：先看显式传入的 ``runtime.config``，再回落到运行上下文。
- **两个来源都要试，不能只看第一个。** 摘要压缩中间件在模型调用里需要后端时按框架
- 自己的写法构造 runtime：
- config = cast("RunnableConfig", getattr(runtime, "config", {}))
- 而那里的 ``request.runtime`` 是 ``langgraph.runtime.Runtime``——**它没有 ``config``
- 字段**，于是 config 恒为 ``{}``（框架注释也明说空字典是合法取值）。若只判断"属性
- 是否存在"，就会把 ``{}`` 当成有效配置、以"缺少会话身份"炸掉整轮——表现为长会话或
- 读了长文档的子智能体**一触发摘要压缩就失败**，而不是压缩生效。
- """
- from langgraph.config import get_config
- candidates: list[Any] = [getattr(runtime, "config", None)]
- try:
- candidates.append(get_config())
- except RuntimeError:
- # 不在 runnable 上下文里（脚本或测试直接调工厂）时没有运行配置可言
- pass
- for candidate in candidates:
- configurable = (candidate or {}).get("configurable") or {}
- user = configurable.get("workspace_user")
- session = configurable.get("workspace_session")
- if user and session:
- return str(user), str(session)
- raise ValueError("文件后端缺少服务端会话身份")
- def backend_for_runtime(runtime):
- from deepagents.backends import CompositeBackend, FilesystemBackend
- user, session = _session_identity(runtime)
- root = session_root(user, session)
- for directory in ("workspace", "conversation_history"):
- (root / directory).mkdir(exist_ok=True)
- return CompositeBackend(
- default=FilesystemBackend(root_dir=root, virtual_mode=True),
- routes={"/skills/": FilesystemBackend(root_dir=config.ROOT / "skills", virtual_mode=True)},
- )
|
