b523b4fd21
- 薄自建 LLM 网关：OpenAI 兼容适配器(DeepSeek) + instructor 结构化输出 + usage_ledger 记账 + 档位路由 - 记忆服务 assemble：确定性选择(显式+主角+近况) + 渲染卡 + 缓存断点(中性文本) - LangGraph 写章节点 + Postgres checkpointer + SSE 归一(token/done/error) - API：立项 + 写章 draft(SSE) + PUT 自动保存 + 提供商凭据(Fernet 加密/测试连接) - 前端：AppShell + 作品库 + 5 步立项向导 + 写作工作台(流式打字机+自动保存) + 设置页 - M1 E2E：真实 DB + mock 网关零 token 走通闭环
57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
"""内存替身：凭据存储 + 提供商探测（端点测试用，无 DB/无网络）。
|
||
|
||
绝对导入 `from fakes import ...`（包目录无 __init__.py，见 memory/gotchas）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
|
||
from ww_api.services.credentials import StoredCredential, StoredRouting
|
||
from ww_llm_gateway.adapters.base import Capabilities
|
||
|
||
|
||
class FakeCredentialStore:
|
||
"""实现 `CredentialStore` Protocol 的内存版。"""
|
||
|
||
def __init__(self) -> None:
|
||
# key: (owner_id, provider) → 密文
|
||
self.creds: dict[tuple[uuid.UUID, str], bytes] = {}
|
||
self.routing: dict[str, StoredRouting] = {}
|
||
|
||
async def list_credentials(self, owner_id: uuid.UUID) -> list[StoredCredential]:
|
||
return [
|
||
StoredCredential(provider=p, api_key_enc=blob)
|
||
for (o, p), blob in self.creds.items()
|
||
if o == owner_id
|
||
]
|
||
|
||
async def list_routing(self) -> list[StoredRouting]:
|
||
return list(self.routing.values())
|
||
|
||
async def get_credential(self, owner_id: uuid.UUID, provider: str) -> StoredCredential | None:
|
||
blob = self.creds.get((owner_id, provider))
|
||
if blob is None:
|
||
return None
|
||
return StoredCredential(provider=provider, api_key_enc=blob)
|
||
|
||
async def upsert_credential(
|
||
self, owner_id: uuid.UUID, provider: str, api_key_enc: bytes
|
||
) -> None:
|
||
self.creds[(owner_id, provider)] = api_key_enc
|
||
|
||
async def upsert_routing(self, routing: StoredRouting) -> None:
|
||
self.routing[routing.tier] = routing
|
||
|
||
|
||
class FakeProviderProbe:
|
||
"""实现 `ProviderProbe` Protocol：返回固定能力矩阵，绝不联网。"""
|
||
|
||
def __init__(self, caps: Capabilities | None = None) -> None:
|
||
self.caps = caps or Capabilities(structured_output=True, prefix_cache=True, thinking=False)
|
||
self.calls: list[tuple[uuid.UUID, str]] = []
|
||
|
||
async def probe(self, owner_id: uuid.UUID, provider: str) -> Capabilities:
|
||
self.calls.append((owner_id, provider))
|
||
return self.caps
|