--- layout: default title: "Python API 重构技术规范" parent: "Chinese (Beta)" --- # Python API 重构技术规范 > **Beta Translation:** This document was translated via Machine Learning and as such may not be 100% accurate. All non-English languages are currently classified as Beta. ## 概述 本规范描述了 TrustGraph Python API 客户端库的全面重构，旨在实现与 API 网关的功能对齐，并添加对现代实时通信模式的支持。 该重构解决了四个主要用例： 1. **流式 LLM 交互**: 启用 LLM 响应的实时流式传输（代理、图 RAG、文档 RAG、文本补全、提示），延迟降低约 60 倍（第一个 token 为 500 毫秒，而不是 30 秒）。 2. **批量数据操作**: 支持高效的大规模知识图谱管理的 triples、图嵌入和文档嵌入的批量导入/导出。 3. **功能对齐**: 确保每个 API 网关端点都有相应的 Python API 方法，包括图嵌入查询。 4. **持久连接**: 启用基于 WebSocket 的通信，以实现多路复用请求和降低连接开销。 ## 目标 **功能对齐**: 每个网关 API 服务都有相应的 Python API 方法。 **流式支持**: 所有支持流式传输的服务（代理、RAG、文本补全、提示）在 Python API 中支持流式传输。 **WebSocket 传输**: 添加可选的 WebSocket 传输层，用于持久连接和多路复用。 **批量操作**: 添加 triples、图嵌入和文档嵌入的高效批量导入/导出功能。 **完全异步支持**: 为所有接口（REST、WebSocket、批量操作、指标）实现完整的 async/await。 **向后兼容性**: 现有代码无需修改即可继续工作。 **类型安全**: 维护类型安全的接口，使用 dataclasses 和类型提示。 **渐进式增强**: 流式传输和异步是可选的，通过显式接口选择启用。 **性能**: 实现流式操作的 60 倍延迟改进。 **现代 Python**: 支持同步和异步范式，以实现最大的灵活性。 ## 背景 ### 当前状态 Python API (`trustgraph-base/trustgraph/api/`) 是一个仅支持 REST 的客户端库，包含以下模块： `flow.py`: 流管理和流范围服务（50 个方法）。 `library.py`: 文档库操作（9 个方法）。 `knowledge.py`: KG 核心管理（4 个方法）。 `collection.py`: 集合元数据（3 个方法）。 `config.py`: 配置管理（6 个方法）。 `types.py`: 数据类型定义（5 个 dataclasses）。 **总操作**: 50/59 (覆盖率 85%) ### 当前限制 **缺失的操作**: 图嵌入查询（图实体上的语义搜索）。 triples、图嵌入、文档嵌入、实体上下文、对象的批量导入/导出。 指标端点。 **缺失的功能**: LLM 服务的流式传输支持。 WebSocket 传输。 多路复用的并发请求。 持久连接。 **性能问题**: LLM 交互的延迟高（第一个 token 需要 30 秒）。 批量数据传输效率低下（每个项目一个 REST 请求）。 多个连续操作的连接开销。 **用户体验问题**: LLM 生成过程中没有实时反馈。 无法取消长时间运行的 LLM 操作。 批量操作的可扩展性差。 ### 影响 2024 年 11 月对网关 API 的流式增强，将 LLM 交互的延迟提高了 60 倍（第一个 token 为 500 毫秒，而不是 30 秒），但 Python API 用户无法利用此功能。 这在 Python 用户和非 Python 用户之间造成了显著的体验差距。 ## 技术设计 ### 架构 重构的 Python API 采用 **模块化接口方法**，具有用于不同通信模式的单独对象。 所有接口都提供 **同步和异步** 两种版本： 1. **REST 接口** (现有，增强) **同步**: `api.flow()`, `api.library()`, `api.knowledge()`, `api.collection()`, `api.config()` **异步**: `api.async_flow()` 同步/异步请求/响应。 简单的连接模型。 默认用于向后兼容。 2. **WebSocket 接口** (新) **同步**: `api.socket()` **异步**: `api.async_socket()` 持久连接。 多路复用的请求。 流式传输支持。 功能重叠时，方法签名与 REST 相同。 3. **批量操作接口** (新) **同步**: `api.bulk()` **异步**: `api.async_bulk()` 基于 WebSocket，提高效率。 基于 Iterator/AsyncIterator 的导入/导出。 处理大型数据集。 4. **指标接口** (新) **同步**: `api.metrics()` **异步**: `api.async_metrics()` Prometheus 指标访问。 ```python import asyncio # Synchronous interfaces api = Api(url="http://localhost:8088/") # REST (existing, unchanged) flow = api.flow().id("default") response = flow.agent(question="...", user="...") # WebSocket (new) socket_flow = api.socket().flow("default") response = socket_flow.agent(question="...", user="...") for chunk in socket_flow.agent(question="...", user="...", streaming=True): print(chunk) # Bulk operations (new) bulk = api.bulk() bulk.import_triples(flow="default", triples=triple_generator()) # Asynchronous interfaces async def main(): api = Api(url="http://localhost:8088/") # Async REST (new) flow = api.async_flow().id("default") response = await flow.agent(question="...", user="...") # Async WebSocket (new) socket_flow = api.async_socket().flow("default") async for chunk in socket_flow.agent(question="...", streaming=True): print(chunk) # Async bulk operations (new) bulk = api.async_bulk() await bulk.import_triples(flow="default", triples=async_triple_generator()) asyncio.run(main()) ``` **关键设计原则**: **所有接口使用相同的URL**: `Api(url="http://localhost:8088/")` 对所有接口都有效。 **同步/异步对称**: 每个接口都具有同步和异步变体，方法签名相同。 **相同的签名**: 当功能重叠时，REST 和 WebSocket 之间的同步和异步方法签名相同。 **渐进式增强**: 根据需求选择接口（REST 用于简单操作，WebSocket 用于流式传输，Bulk 用于大型数据集，async 用于现代框架）。 **明确意图**: `api.socket()` 表示 WebSocket，`api.async_socket()` 表示异步 WebSocket。 **向后兼容**: 现有代码保持不变。 ### 组件 #### 1. 核心 API 类 (修改版) 模块: `trustgraph-base/trustgraph/api/api.py` **增强的 API 类**: ```python class Api: def __init__(self, url: str, timeout: int = 60, token: Optional[str] = None): self.url = url self.timeout = timeout self.token = token # Optional bearer token for REST, query param for WebSocket self._socket_client = None self._bulk_client = None self._async_flow = None self._async_socket_client = None self._async_bulk_client = None # Existing synchronous methods (unchanged) def flow(self) -> Flow: """Synchronous REST-based flow interface""" pass def library(self) -> Library: """Synchronous REST-based library interface""" pass def knowledge(self) -> Knowledge: """Synchronous REST-based knowledge interface""" pass def collection(self) -> Collection: """Synchronous REST-based collection interface""" pass def config(self) -> Config: """Synchronous REST-based config interface""" pass # New synchronous methods def socket(self) -> SocketClient: """Synchronous WebSocket-based interface for streaming operations""" if self._socket_client is None: self._socket_client = SocketClient(self.url, self.timeout, self.token) return self._socket_client def bulk(self) -> BulkClient: """Synchronous bulk operations interface for import/export""" if self._bulk_client is None: self._bulk_client = BulkClient(self.url, self.timeout, self.token) return self._bulk_client def metrics(self) -> Metrics: """Synchronous metrics interface""" return Metrics(self.url, self.timeout, self.token) # New asynchronous methods def async_flow(self) -> AsyncFlow: """Asynchronous REST-based flow interface""" if self._async_flow is None: self._async_flow = AsyncFlow(self.url, self.timeout, self.token) return self._async_flow def async_socket(self) -> AsyncSocketClient: """Asynchronous WebSocket-based interface for streaming operations""" if self._async_socket_client is None: self._async_socket_client = AsyncSocketClient(self.url, self.timeout, self.token) return self._async_socket_client def async_bulk(self) -> AsyncBulkClient: """Asynchronous bulk operations interface for import/export""" if self._async_bulk_client is None: self._async_bulk_client = AsyncBulkClient(self.url, self.timeout, self.token) return self._async_bulk_client def async_metrics(self) -> AsyncMetrics: """Asynchronous metrics interface""" return AsyncMetrics(self.url, self.timeout, self.token) # Resource management def close(self) -> None: """Close all synchronous connections""" if self._socket_client: self._socket_client.close() if self._bulk_client: self._bulk_client.close() async def aclose(self) -> None: """Close all asynchronous connections""" if self._async_socket_client: await self._async_socket_client.aclose() if self._async_bulk_client: await self._async_bulk_client.aclose() if self._async_flow: await self._async_flow.aclose() def __enter__(self): return self def __exit__(self, *args): self.close() async def __aenter__(self): return self async def __aexit__(self, *args): await self.aclose() ``` #### 2. 同步 WebSocket 客户端 模块：`trustgraph-base/trustgraph/api/socket_client.py` (新) **SocketClient 类**: ```python class SocketClient: """Synchronous WebSocket client""" def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = self._convert_to_ws_url(url) self.timeout = timeout self.token = token self._connection = None self._request_counter = 0 def flow(self, flow_id: str) -> SocketFlowInstance: """Get flow instance for WebSocket operations""" return SocketFlowInstance(self, flow_id) def _connect(self) -> WebSocket: """Establish WebSocket connection (lazy)""" # Uses asyncio.run() internally to wrap async websockets library pass def _send_request( self, service: str, flow: Optional[str], request: Dict[str, Any], streaming: bool = False ) -> Union[Dict[str, Any], Iterator[Dict[str, Any]]]: """Send request and handle response/streaming""" # Synchronous wrapper around async WebSocket calls pass def close(self) -> None: """Close WebSocket connection""" pass class SocketFlowInstance: """Synchronous WebSocket flow instance with same interface as REST FlowInstance""" def __init__(self, client: SocketClient, flow_id: str): self.client = client self.flow_id = flow_id # Same method signatures as FlowInstance def agent( self, question: str, user: str, state: Optional[Dict[str, Any]] = None, group: Optional[str] = None, history: Optional[List[Dict[str, Any]]] = None, streaming: bool = False, **kwargs ) -> Union[Dict[str, Any], Iterator[Dict[str, Any]]]: """Agent with optional streaming""" pass def text_completion( self, system: str, prompt: str, streaming: bool = False, **kwargs ) -> Union[str, Iterator[str]]: """Text completion with optional streaming""" pass # ... similar for graph_rag, document_rag, prompt, etc. ``` **主要特性：** 延迟连接（仅在首次发送请求时建立连接） 请求复用（最多 15 个并发请求） 断开连接时自动重连 流式响应解析 线程安全操作 异步 WebSocket 库的同步封装 #### 3. 异步 WebSocket 客户端 模块：`trustgraph-base/trustgraph/api/async_socket_client.py` (新) **AsyncSocketClient 类：** ```python class AsyncSocketClient: """Asynchronous WebSocket client""" def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = self._convert_to_ws_url(url) self.timeout = timeout self.token = token self._connection = None self._request_counter = 0 def flow(self, flow_id: str) -> AsyncSocketFlowInstance: """Get async flow instance for WebSocket operations""" return AsyncSocketFlowInstance(self, flow_id) async def _connect(self) -> WebSocket: """Establish WebSocket connection (lazy)""" # Native async websockets library pass async def _send_request( self, service: str, flow: Optional[str], request: Dict[str, Any], streaming: bool = False ) -> Union[Dict[str, Any], AsyncIterator[Dict[str, Any]]]: """Send request and handle response/streaming""" pass async def aclose(self) -> None: """Close WebSocket connection""" pass class AsyncSocketFlowInstance: """Asynchronous WebSocket flow instance""" def __init__(self, client: AsyncSocketClient, flow_id: str): self.client = client self.flow_id = flow_id # Same method signatures as FlowInstance (but async) async def agent( self, question: str, user: str, state: Optional[Dict[str, Any]] = None, group: Optional[str] = None, history: Optional[List[Dict[str, Any]]] = None, streaming: bool = False, **kwargs ) -> Union[Dict[str, Any], AsyncIterator[Dict[str, Any]]]: """Agent with optional streaming""" pass async def text_completion( self, system: str, prompt: str, streaming: bool = False, **kwargs ) -> Union[str, AsyncIterator[str]]: """Text completion with optional streaming""" pass # ... similar for graph_rag, document_rag, prompt, etc. ``` **主要特性：** 原生支持 async/await 适用于异步应用程序（FastAPI, aiohttp） 无线程阻塞 与同步版本具有相同的接口 AsyncIterator 用于流式传输 #### 4. 同步批量操作客户端 模块：`trustgraph-base/trustgraph/api/bulk_client.py` (新) **BulkClient 类：** ```python class BulkClient: """Synchronous bulk operations client""" def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = self._convert_to_ws_url(url) self.timeout = timeout self.token = token def import_triples( self, flow: str, triples: Iterator[Triple], **kwargs ) -> None: """Bulk import triples via WebSocket""" pass def export_triples( self, flow: str, **kwargs ) -> Iterator[Triple]: """Bulk export triples via WebSocket""" pass def import_graph_embeddings( self, flow: str, embeddings: Iterator[Dict[str, Any]], **kwargs ) -> None: """Bulk import graph embeddings via WebSocket""" pass def export_graph_embeddings( self, flow: str, **kwargs ) -> Iterator[Dict[str, Any]]: """Bulk export graph embeddings via WebSocket""" pass # ... similar for document embeddings, entity contexts, objects def close(self) -> None: """Close connections""" pass ``` **主要特性：** 基于迭代器，内存占用恒定 每个操作都使用专用的 WebSocket 连接 进度跟踪（可选的回调函数） 错误处理，支持部分成功报告 #### 5. 异步批量操作客户端 模块：`trustgraph-base/trustgraph/api/async_bulk_client.py` (新) **AsyncBulkClient 类：** ```python class AsyncBulkClient: """Asynchronous bulk operations client""" def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = self._convert_to_ws_url(url) self.timeout = timeout self.token = token async def import_triples( self, flow: str, triples: AsyncIterator[Triple], **kwargs ) -> None: """Bulk import triples via WebSocket""" pass async def export_triples( self, flow: str, **kwargs ) -> AsyncIterator[Triple]: """Bulk export triples via WebSocket""" pass async def import_graph_embeddings( self, flow: str, embeddings: AsyncIterator[Dict[str, Any]], **kwargs ) -> None: """Bulk import graph embeddings via WebSocket""" pass async def export_graph_embeddings( self, flow: str, **kwargs ) -> AsyncIterator[Dict[str, Any]]: """Bulk export graph embeddings via WebSocket""" pass # ... similar for document embeddings, entity contexts, objects async def aclose(self) -> None: """Close connections""" pass ``` **主要特性：** 基于 AsyncIterator，内存占用恒定 适用于异步应用程序，效率高 原生支持 async/await 与同步版本具有相同的接口 #### 6. REST Flow API (同步 - 未改变) 模块：`trustgraph-base/trustgraph/api/flow.py` REST Flow API 保持 **完全不变**，以确保向后兼容。所有现有方法均可继续使用： `Flow.list()`, `Flow.start()`, `Flow.stop()`, 等等。 `FlowInstance.agent()`, `FlowInstance.text_completion()`, `FlowInstance.graph_rag()`, 等等。 所有现有签名和返回类型均已保留 **新增：** 为了实现功能对等，向 REST FlowInstance 添加 `graph_embeddings_query()`： ```python class FlowInstance: # All existing methods unchanged... # New: Graph embeddings query (REST) def graph_embeddings_query( self, text: str, user: str, collection: str, limit: int = 10, **kwargs ) -> List[Dict[str, Any]]: """Query graph embeddings for semantic search""" # Calls POST /api/v1/flow/{flow}/service/graph-embeddings pass ``` #### 7. 异步 REST 流程 API 模块：`trustgraph-base/trustgraph/api/async_flow.py` (新) **AsyncFlow 和 AsyncFlowInstance 类：** ```python class AsyncFlow: """Asynchronous REST-based flow interface""" def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = url self.timeout = timeout self.token = token async def list(self) -> List[Dict[str, Any]]: """List all flows""" pass async def get(self, id: str) -> Dict[str, Any]: """Get flow definition""" pass async def start(self, class_name: str, id: str, description: str, parameters: Dict) -> None: """Start a flow""" pass async def stop(self, id: str) -> None: """Stop a flow""" pass def id(self, flow_id: str) -> AsyncFlowInstance: """Get async flow instance""" return AsyncFlowInstance(self.url, self.timeout, self.token, flow_id) async def aclose(self) -> None: """Close connection""" pass class AsyncFlowInstance: """Asynchronous REST flow instance""" async def agent( self, question: str, user: str, state: Optional[Dict[str, Any]] = None, group: Optional[str] = None, history: Optional[List[Dict[str, Any]]] = None, **kwargs ) -> Dict[str, Any]: """Async agent execution""" pass async def text_completion( self, system: str, prompt: str, **kwargs ) -> str: """Async text completion""" pass async def graph_rag( self, question: str, user: str, collection: str, **kwargs ) -> str: """Async graph RAG""" pass # ... all other FlowInstance methods as async versions ``` **主要特性：** 使用 `aiohttp` 或 `httpx` 实现原生异步 HTTP。 方法签名与同步 REST API 相同。 不支持流式传输（使用 `async_socket()` 进行流式传输）。 适用于异步应用程序。 #### 8. Metrics API 模块：`trustgraph-base/trustgraph/api/metrics.py` (新) **同步 Metrics：** ```python class Metrics: def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = url self.timeout = timeout self.token = token def get(self) -> str: """Get Prometheus metrics as text""" # Call GET /api/metrics pass ``` **异步指标：** ```python class AsyncMetrics: def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = url self.timeout = timeout self.token = token async def get(self) -> str: """Get Prometheus metrics as text""" # Call GET /api/metrics pass ``` #### 9. 增强的类型 模块：`trustgraph-base/trustgraph/api/types.py` (已修改) **新类型：** ```python from typing import Iterator, Union, Dict, Any import dataclasses @dataclasses.dataclass class StreamingChunk: """Base class for streaming chunks""" content: str end_of_message: bool = False @dataclasses.dataclass class AgentThought(StreamingChunk): """Agent reasoning chunk""" chunk_type: str = "thought" @dataclasses.dataclass class AgentObservation(StreamingChunk): """Agent tool observation chunk""" chunk_type: str = "observation" @dataclasses.dataclass class AgentAnswer(StreamingChunk): """Agent final answer chunk""" chunk_type: str = "final-answer" end_of_dialog: bool = False @dataclasses.dataclass class RAGChunk(StreamingChunk): """RAG streaming chunk""" end_of_stream: bool = False error: Optional[Dict[str, str]] = None # Type aliases for clarity AgentStream = Iterator[Union[AgentThought, AgentObservation, AgentAnswer]] RAGStream = Iterator[RAGChunk] CompletionStream = Iterator[str] ``` #### 6. 指标 API 模块：`trustgraph-base/trustgraph/api/metrics.py` (新) ```python class Metrics: def __init__(self, url: str, timeout: int, token: Optional[str]): self.url = url self.timeout = timeout self.token = token def get(self) -> str: """Get Prometheus metrics as text""" # Call GET /api/metrics pass ``` ### 实现方案 #### 第一阶段：核心 API 增强 (第一周) 1. 向 `Api` 类添加 `socket()`、`bulk()` 和 `metrics()` 方法 2. 实现 WebSocket 和批量客户端的延迟初始化 3. 添加上下文管理器支持 (`__enter__`, `__exit__`) 4. 添加 `close()` 方法进行清理 5. 为 API 类增强添加单元测试 6. 验证向后兼容性 **向后兼容性**: 没有破坏性更改。仅添加了新方法。 #### 第二阶段：WebSocket 客户端 (第二周 - 第三周) 1. 实现 `SocketClient` 类，用于连接管理 2. 实现 `SocketFlowInstance`，方法签名与 `FlowInstance` 相同 3. 添加请求复用支持 (最多 15 个并发) 4. 添加流式响应解析，支持不同的分块类型 5. 添加自动重连逻辑 6. 添加单元和集成测试 7. 记录 WebSocket 使用模式 **向后兼容性**: 仅为新接口。对现有代码没有影响。 #### 第三阶段：流式支持 (第三周 - 第四周) 1. 添加流式分块类型类 (`AgentThought`, `AgentObservation`, `AgentAnswer`, `RAGChunk`) 2. 在 `SocketClient` 中实现流式响应解析 3. 将流式参数添加到 `SocketFlowInstance` 中的所有 LLM 方法 4. 处理流式过程中的错误情况 5. 添加流式功能的单元和集成测试 6. 在文档中添加流式示例 **向后兼容性**: 仅为新接口。现有 REST API 未改变。 #### 第四阶段：批量操作 (第四周 - 第五周) 1. 实现 `BulkClient` 类 2. 添加三元组、嵌入、上下文、对象的批量导入/导出方法 3. 实现基于迭代器的处理，以实现恒定内存 4. 添加进度跟踪 (可选的回调) 5. 添加错误处理，并报告部分成功 6. 添加单元和集成测试 7. 添加批量操作示例 **向后兼容性**: 仅为新接口。对现有代码没有影响。 #### 第五阶段：功能完善与优化 (第五周) 1. 向 REST `FlowInstance` 添加 `graph_embeddings_query()` 2. 实现 `Metrics` 类 3. 添加全面的集成测试 4. 性能基准测试 5. 更新所有文档 6. 创建迁移指南 **向后兼容性**: 仅添加新方法。对现有代码没有影响。 ### 数据模型 #### 接口选择 ```python # Single API instance, same URL for all interfaces api = Api(url="http://localhost:8088/") # Synchronous interfaces rest_flow = api.flow().id("default") # Sync REST socket_flow = api.socket().flow("default") # Sync WebSocket bulk = api.bulk() # Sync bulk operations metrics = api.metrics() # Sync metrics # Asynchronous interfaces async_rest_flow = api.async_flow().id("default") # Async REST async_socket_flow = api.async_socket().flow("default") # Async WebSocket async_bulk = api.async_bulk() # Async bulk operations async_metrics = api.async_metrics() # Async metrics ``` #### 流式响应类型 **代理流式传输**: ```python api = Api(url="http://localhost:8088/") # REST interface - non-streaming (existing) rest_flow = api.flow().id("default") response = rest_flow.agent(question="What is ML?", user="user123") print(response["response"]) # WebSocket interface - non-streaming (same signature) socket_flow = api.socket().flow("default") response = socket_flow.agent(question="What is ML?", user="user123") print(response["response"]) # WebSocket interface - streaming (new) for chunk in socket_flow.agent(question="What is ML?", user="user123", streaming=True): if isinstance(chunk, AgentThought): print(f"Thinking: {chunk.content}") elif isinstance(chunk, AgentObservation): print(f"Observed: {chunk.content}") elif isinstance(chunk, AgentAnswer): print(f"Answer: {chunk.content}") if chunk.end_of_dialog: break ``` **RAG 流式传输**: ```python api = Api(url="http://localhost:8088/") # REST interface - non-streaming (existing) rest_flow = api.flow().id("default") response = rest_flow.graph_rag(question="What is Python?", user="user123", collection="default") print(response) # WebSocket interface - streaming (new) socket_flow = api.socket().flow("default") for chunk in socket_flow.graph_rag( question="What is Python?", user="user123", collection="default", streaming=True ): print(chunk.content, end="", flush=True) if chunk.end_of_stream: break ``` **批量操作 (同步)**: ```python api = Api(url="http://localhost:8088/") # Bulk import triples def triple_generator(): yield Triple(s="http://ex.com/alice", p="http://ex.com/type", o="Person") yield Triple(s="http://ex.com/alice", p="http://ex.com/name", o="Alice") yield Triple(s="http://ex.com/bob", p="http://ex.com/type", o="Person") bulk = api.bulk() bulk.import_triples(flow="default", triples=triple_generator()) # Bulk export triples for triple in bulk.export_triples(flow="default"): print(f"{triple.s} -> {triple.p} -> {triple.o}") ``` **批量操作（异步）**: ```python import asyncio async def main(): api = Api(url="http://localhost:8088/") # Async bulk import triples async def async_triple_generator(): yield Triple(s="http://ex.com/alice", p="http://ex.com/type", o="Person") yield Triple(s="http://ex.com/alice", p="http://ex.com/name", o="Alice") yield Triple(s="http://ex.com/bob", p="http://ex.com/type", o="Person") bulk = api.async_bulk() await bulk.import_triples(flow="default", triples=async_triple_generator()) # Async bulk export triples async for triple in bulk.export_triples(flow="default"): print(f"{triple.s} -> {triple.p} -> {triple.o}") asyncio.run(main()) ``` **异步 REST 示例：** ```python import asyncio async def main(): api = Api(url="http://localhost:8088/") # Async REST flow operations flow = api.async_flow().id("default") response = await flow.agent(question="What is ML?", user="user123") print(response["response"]) asyncio.run(main()) ``` **异步 WebSocket 流式传输示例：** ```python import asyncio async def main(): api = Api(url="http://localhost:8088/") # Async WebSocket streaming socket = api.async_socket() flow = socket.flow("default") async for chunk in flow.agent(question="What is ML?", user="user123", streaming=True): if isinstance(chunk, AgentAnswer): print(chunk.content, end="", flush=True) if chunk.end_of_dialog: break asyncio.run(main()) ``` ### APIs #### New APIs 1. **Core API Class**: **Synchronous**: `Api.socket()` - Get synchronous WebSocket client `Api.bulk()` - Get synchronous bulk operations client `Api.metrics()` - Get synchronous metrics client `Api.close()` - Close all synchronous connections Context manager support (`__enter__`, `__exit__`) **Asynchronous**: `Api.async_flow()` - Get asynchronous REST flow client `Api.async_socket()` - Get asynchronous WebSocket client `Api.async_bulk()` - Get asynchronous bulk operations client `Api.async_metrics()` - Get asynchronous metrics client `Api.aclose()` - Close all asynchronous connections Async context manager support (`__aenter__`, `__aexit__`) 2. **Synchronous WebSocket Client**: `SocketClient.flow(flow_id)` - Get WebSocket flow instance `SocketFlowInstance.agent(..., streaming: bool = False)` - Agent with optional streaming `SocketFlowInstance.text_completion(..., streaming: bool = False)` - Text completion with optional streaming `SocketFlowInstance.graph_rag(..., streaming: bool = False)` - Graph RAG with optional streaming `SocketFlowInstance.document_rag(..., streaming: bool = False)` - Document RAG with optional streaming `SocketFlowInstance.prompt(..., streaming: bool = False)` - Prompt with optional streaming `SocketFlowInstance.graph_embeddings_query()` - Graph embeddings query All other FlowInstance methods with identical signatures 3. **Asynchronous WebSocket Client**: `AsyncSocketClient.flow(flow_id)` - Get async WebSocket flow instance `AsyncSocketFlowInstance.agent(..., streaming: bool = False)` - Async agent with optional streaming `AsyncSocketFlowInstance.text_completion(..., streaming: bool = False)` - Async text completion with optional streaming `AsyncSocketFlowInstance.graph_rag(..., streaming: bool = False)` - Async graph RAG with optional streaming `AsyncSocketFlowInstance.document_rag(..., streaming: bool = False)` - Async document RAG with optional streaming `AsyncSocketFlowInstance.prompt(..., streaming: bool = False)` - Async prompt with optional streaming `AsyncSocketFlowInstance.graph_embeddings_query()` - Async graph embeddings query All other FlowInstance methods as async versions 4. **Synchronous Bulk Operations Client**: `BulkClient.import_triples(flow, triples)` - Bulk triple import `BulkClient.export_triples(flow)` - Bulk triple export `BulkClient.import_graph_embeddings(flow, embeddings)` - Bulk graph embeddings import `BulkClient.export_graph_embeddings(flow)` - Bulk graph embeddings export `BulkClient.import_document_embeddings(flow, embeddings)` - Bulk document embeddings import `BulkClient.export_document_embeddings(flow)` - Bulk document embeddings export `BulkClient.import_entity_contexts(flow, contexts)` - Bulk entity contexts import `BulkClient.export_entity_contexts(flow)` - Bulk entity contexts export `BulkClient.import_objects(flow, objects)` - Bulk objects import 5. **Asynchronous Bulk Operations Client**: `AsyncBulkClient.import_triples(flow, triples)` - Async bulk triple import `AsyncBulkClient.export_triples(flow)` - Async bulk triple export `AsyncBulkClient.import_graph_embeddings(flow, embeddings)` - Async bulk graph embeddings import `AsyncBulkClient.export_graph_embeddings(flow)` - Async bulk graph embeddings export `AsyncBulkClient.import_document_embeddings(flow, embeddings)` - Async bulk document embeddings import `AsyncBulkClient.export_document_embeddings(flow)` - Async bulk document embeddings export `AsyncBulkClient.import_entity_contexts(flow, contexts)` - Async bulk entity contexts import `AsyncBulkClient.export_entity_contexts(flow)` - Async bulk entity contexts export `AsyncBulkClient.import_objects(flow, objects)` - Async bulk objects import 6. **Asynchronous REST Flow Client**: `AsyncFlow.list()` - Async list all flows `AsyncFlow.get(id)` - Async get flow definition `AsyncFlow.start(...)` - Async start flow `AsyncFlow.stop(id)` - Async stop flow `AsyncFlow.id(flow_id)` - Get async flow instance `AsyncFlowInstance.agent(...)` - Async agent execution `AsyncFlowInstance.text_completion(...)` - Async text completion `AsyncFlowInstance.graph_rag(...)` - Async graph RAG All other FlowInstance methods as async versions 7. **Metrics Clients**: `Metrics.get()` - Synchronous Prometheus metrics `AsyncMetrics.get()` - Asynchronous Prometheus metrics 8. **REST Flow API Enhancement**: `FlowInstance.graph_embeddings_query()` - Graph embeddings query (sync feature parity) `AsyncFlowInstance.graph_embeddings_query()` - Graph embeddings query (async feature parity) #### Modified APIs 1. **Constructor** (minor enhancement): ```python Api(url: str, timeout: int = 60, token: Optional[str] = None) ``` 添加了 `token` 参数（可选，用于身份验证） 如果未指定 `None`（默认）：未使用身份验证 如果已指定：用作 REST 的 bearer token（`Authorization: Bearer `），用作 WebSocket 的查询参数（`?token=`） 没有其他更改 - 完全向后兼容 2. **无破坏性更改：** 所有现有的 REST API 方法未更改 所有现有的签名均保留 所有现有的返回类型均保留 ### 实现细节 #### 错误处理 **WebSocket 连接错误：** ```python try: api = Api(url="http://localhost:8088/") socket = api.socket() socket_flow = socket.flow("default") response = socket_flow.agent(question="...", user="user123") except ConnectionError as e: print(f"WebSocket connection failed: {e}") print("Hint: Ensure Gateway is running and WebSocket endpoint is accessible") ``` **优雅的降级方案**: ```python api = Api(url="http://localhost:8088/") try: # Try WebSocket streaming first socket_flow = api.socket().flow("default") for chunk in socket_flow.agent(question="...", user="...", streaming=True): print(chunk.content) except ConnectionError: # Fall back to REST non-streaming print("WebSocket unavailable, falling back to REST") rest_flow = api.flow().id("default") response = rest_flow.agent(question="...", user="...") print(response["response"]) ``` **部分流式错误：** ```python api = Api(url="http://localhost:8088/") socket_flow = api.socket().flow("default") accumulated = [] try: for chunk in socket_flow.graph_rag(question="...", streaming=True): accumulated.append(chunk.content) if chunk.error: print(f"Error occurred: {chunk.error}") print(f"Partial response: {''.join(accumulated)}") break except Exception as e: print(f"Streaming error: {e}") print(f"Partial response: {''.join(accumulated)}") ``` #### 资源管理 **上下文管理器支持：** ```python # Automatic cleanup with Api(url="http://localhost:8088/") as api: socket_flow = api.socket().flow("default") response = socket_flow.agent(question="...", user="user123") # All connections automatically closed # Manual cleanup api = Api(url="http://localhost:8088/") try: socket_flow = api.socket().flow("default") response = socket_flow.agent(question="...", user="user123") finally: api.close() # Explicitly close all connections (WebSocket, bulk, etc.) ``` #### 线程和并发 **线程安全**: 每个 `Api` 实例维护自己的连接。 WebSocket 传输使用锁进行线程安全的请求复用。 多个线程可以安全地共享一个 `Api` 实例。 流式迭代器不具备线程安全特性（只能从单个线程消费）。 **异步支持**（未来考虑）： ```python # Phase 2 enhancement (not in initial scope) import asyncio async def main(): api = await AsyncApi(url="ws://localhost:8088/") flow = api.flow().id("default") async for chunk in flow.agent(question="...", streaming=True): print(chunk.content) await api.close() asyncio.run(main()) ``` ## 安全注意事项 ### 身份验证 **令牌参数**: ```python # No authentication (default) api = Api(url="http://localhost:8088/") # With authentication api = Api(url="http://localhost:8088/", token="mytoken") ``` **REST 传输：** 通过 `Authorization` 头部传递认证令牌 自动应用于所有 REST 请求 格式：`Authorization: Bearer ` **WebSocket 传输：** 通过附加到 WebSocket URL 的查询参数传递认证令牌 在连接建立期间自动应用 格式：`ws://localhost:8088/api/v1/socket?token=` **实现：** ```python class SocketClient: def _connect(self) -> WebSocket: # Construct WebSocket URL with optional token ws_url = f"{self.url}/api/v1/socket" if self.token: ws_url = f"{ws_url}?token={self.token}" # Connect to WebSocket return websocket.connect(ws_url) ``` **示例：** ```python # REST with auth api = Api(url="http://localhost:8088/", token="mytoken") flow = api.flow().id("default") # All REST calls include: Authorization: Bearer mytoken # WebSocket with auth socket = api.socket() # Connects to: ws://localhost:8088/api/v1/socket?token=mytoken ``` ### 安全通信 支持 WS (WebSocket) 和 WSS (WebSocket Secure) 两种协议。 WSS 连接的 TLS 证书验证。 可选的开发环境下的证书验证禁用（带有警告）。 ### 输入验证 验证 URL 协议 (http, https, ws, wss)。 验证传输参数值。 验证流式传输参数组合。 验证批量导入的数据类型。 ## 性能考虑 ### 延迟优化 **流式 LLM 操作**: **首次 token 时间**: ~500 毫秒 (vs ~30 秒 非流式) **改进**: 性能提升 60 倍 **适用范围**: Agent, Graph RAG, Document RAG, 文本补全, Prompt **持久连接**: **连接开销**: 后续请求消除连接开销 **WebSocket 握手**: 一次性成本 (~100 毫秒) **适用范围**: 使用 WebSocket 传输的所有操作 ### 吞吐量优化 **批量操作**: **三元组导入**: ~10,000 三元组/秒 (vs ~100/秒 使用 REST 单个项目) **嵌入向量导入**: ~5,000 嵌入向量/秒 (vs ~50/秒 使用 REST 单个项目) **改进**: 批量操作的吞吐量提升 100 倍 **请求复用**: **并发请求**: 单个连接可支持最多 15 个并发请求 **连接重用**: 并发操作无需连接开销 ### 内存考虑 **流式响应**: 恒定内存使用 (按接收到的块处理) 不缓冲完整的响应 适用于非常长的输出 (>1MB) **批量操作**: 基于迭代器的处理 (恒定内存) 不将整个数据集加载到内存中 适用于包含数百万项的数据集 ### 基准测试 (预期) | 操作 | REST (现有) | WebSocket (流式传输) | 改进 | |-----------|----------------|----------------------|-------------| | Agent (首次 token 时间) | 30 秒 | 0.5 秒 | 60 倍 | | Graph RAG (首次 token 时间) | 25 秒 | 0.5 秒 | 50 倍 | | 导入 10K 三元组 | 100 秒 | 1 秒 | 100 倍 | | 导入 1M 三元组 | 10,000 秒 (2.7 小时) | 100 秒 (1.6 分钟) | 100 倍 | | 10 个并发小请求 | 5 秒 (顺序) | 0.5 秒 (并行) | 10 倍 | ## 测试策略 ### 单元测试 **传输层** (`test_transport.py`): 测试 REST 传输的请求/响应 测试 WebSocket 传输连接 测试 WebSocket 传输重连 测试请求复用 测试流式响应解析 模拟 WebSocket 服务器以进行确定性测试 **API 方法** (`test_flow.py`, `test_library.py`, 等): 使用模拟传输测试新方法 测试流式参数处理 测试批量操作迭代器 测试错误处理 **类型** (`test_types.py`): 测试新的流式分块类型 测试类型序列化/反序列化 ### 集成测试 **端到端 REST** (`test_integration_rest.py`): 测试所有操作，针对真实的网关（REST 模式） 验证向后兼容性 测试错误条件 **端到端 WebSocket** (`test_integration_websocket.py`): 测试所有操作，针对真实的网关（WebSocket 模式） 测试流式操作 测试批量操作 测试并发请求 测试连接恢复 **流式服务** (`test_streaming_integration.py`): 测试代理流式传输（想法、观察、答案） 测试 RAG 流式传输（增量分块） 测试文本补全流式传输（逐个 token） 测试提示流式传输 测试流式传输过程中的错误处理 **批量操作** (`test_bulk_integration.py`): 测试三元组的批量导入/导出（1K、10K、100K 个条目） 测试嵌入的批量导入/导出 测试批量操作期间的内存使用情况 测试进度跟踪 ### 性能测试 **延迟基准测试** (`test_performance_latency.py`): 测量第一个 token 的时间（流式传输与非流式传输） 测量连接开销（REST 与 WebSocket） 与预期基准进行比较 **吞吐量基准测试** (`test_performance_throughput.py`): 测量批量导入吞吐量 测量请求复用效率 与预期基准进行比较 ### 兼容性测试 **向后兼容性** (`test_backward_compatibility.py`): 运行现有测试套件，针对重构的 API 验证没有破坏性更改 测试常见模式的迁移路径 ## 迁移计划 ### 第一阶段：透明迁移（默认） **无需任何代码更改**。 现有代码继续工作： ```python # Existing code works unchanged api = Api(url="http://localhost:8088/") flow = api.flow().id("default") response = flow.agent(question="What is ML?", user="user123") ``` ### 第二阶段：选择加入的流媒体（简单） **使用 `api.socket()` 接口** 启用流媒体： ```python # Before: Non-streaming REST api = Api(url="http://localhost:8088/") rest_flow = api.flow().id("default") response = rest_flow.agent(question="What is ML?", user="user123") print(response["response"]) # After: Streaming WebSocket (same parameters!) api = Api(url="http://localhost:8088/") # Same URL socket_flow = api.socket().flow("default") for chunk in socket_flow.agent(question="What is ML?", user="user123", streaming=True): if isinstance(chunk, AgentAnswer): print(chunk.content, end="", flush=True) ``` **关键点：** REST 和 WebSocket 都使用相同的 URL 相同的函数签名（易于迁移） 只需要添加 `.socket()` 和 `streaming=True` ### 第三阶段：批量操作（新功能） **使用 `api.bulk()` 接口** 处理大型数据集： ```python # Before: Inefficient per-item operations api = Api(url="http://localhost:8088/") flow = api.flow().id("default") for triple in my_large_triple_list: # Slow per-item operations # (no direct bulk insert in REST API) pass # After: Efficient bulk loading api = Api(url="http://localhost:8088/") # Same URL bulk = api.bulk() # This is fast (10,000 triples/second) bulk.import_triples(flow="default", triples=iter(my_large_triple_list)) ``` ### 文档更新 1. **README.md**: 添加流式传输和 WebSocket 示例 2. **API 参考**: 记录所有新的方法和参数 3. **迁移指南**: 启用流式传输的分步指南 4. **示例**: 添加常用模式的示例脚本 5. **性能指南**: 记录预期的性能改进 ### 弃用策略 **无弃用**. 所有现有的 API 均继续支持。 这是一个纯粹的增强。 ## 时间线 ### 第一周：基础 传输抽象层 重构现有的 REST 代码 传输层的单元测试 向后兼容性验证 ### 第二周：WebSocket 传输 WebSocket 传输实现 连接管理和重连 请求复用 单元和集成测试 ### 第三周：流式传输支持 向 LLM 方法添加流式传输参数 实现流式响应解析 添加流式传输块类型 流式传输集成测试 ### 第四周：批量操作 添加批量导入/导出方法 实现基于迭代器的操作 性能测试 批量操作集成测试 ### 第五周：功能对齐与文档 添加图嵌入查询 添加指标 API 综合文档 迁移指南 候选版本 ### 第六周：发布 最终集成测试 性能基准测试 发布文档 社区公告 **总时长**: 6 周 ## 开放问题 ### API 设计问题 1. **异步支持**: ✅ **已解决** - 异步支持已包含在初始版本中 所有接口都具有异步变体：`async_flow()`, `async_socket()`, `async_bulk()`, `async_metrics()` 为同步和异步 API 提供完全的对称性 对于现代异步框架（FastAPI, aiohttp）至关重要 2. **进度跟踪**: 批量操作是否应支持进度回调？ ```python def progress_callback(processed: int, total: Optional[int]): print(f"Processed {processed} items") bulk.import_triples(flow="default", triples=triples, on_progress=progress_callback) ``` **建议**: 在第二阶段添加。对于初始发布，这不是关键。 3. **流式传输超时**: 我们应该如何处理流式传输操作的超时？ **建议**: 使用与非流式传输相同的超时时间，但在接收每个数据块时重置。 4. **数据块缓冲**: 我们应该缓冲数据块，还是立即返回？ **建议**: 立即返回，以获得最低延迟。 5. **WebSocket 上的全局服务**: `api.socket()` 是否应该支持全局服务（库、知识、集合、配置），或者仅支持作用域服务？ **建议**: 仅从作用域服务开始（流式传输相关）。如果需要，可以在第二阶段添加全局服务。 ### 实现问题 1. **WebSocket 库**: 我们应该使用 `websockets`、`websocket-client` 还是 `aiohttp`？ **建议**: `websockets` (异步，成熟，维护良好)。使用 `asyncio.run()` 包装成同步接口。 2. **连接池**: 我们是否应该支持多个并发的 `Api` 实例共享连接池？ **建议**: 延期到第二阶段。最初，每个 `Api` 实例都有自己的连接。 3. **连接重用**: `SocketClient` 和 `BulkClient` 是否应该共享同一个 WebSocket 连接，或者使用单独的连接？ **建议**: 使用单独的连接。实现更简单，职责分离更清晰。 4. **延迟连接 vs 立即连接**: WebSocket 连接是在 `api.socket()` 中建立，还是在首次请求时建立？ **建议**: 延迟到首次请求时建立。避免连接开销，如果用户仅使用 REST 方法。 ### 测试问题 1. **模拟网关**: 我们是否应该创建一个轻量级的模拟网关进行测试，或者直接测试真实网关？ **建议**: 都应该。使用模拟进行单元测试，使用真实网关进行集成测试。 2. **性能回归测试**: 我们是否应该将自动化性能回归测试添加到 CI 中？ **建议**: 是的，但应设置宽松的阈值，以考虑 CI 环境的差异。 ## 引用 ### 相关技术规范 `docs/tech-specs/streaming-llm-responses.md` - 网关中的流式传输实现 `docs/tech-specs/rag-streaming-support.md` - RAG 流式传输支持 ### 实现文件 `trustgraph-base/trustgraph/api/` - Python API 源代码 `trustgraph-flow/trustgraph/gateway/` - 网关源代码 `trustgraph-flow/trustgraph/gateway/dispatch/mux.py` - WebSocket 多路复用参考实现 ### 文档 `docs/apiSpecification.md` - 完整的 API 参考 `docs/api-status-summary.md` - API 状态摘要 `README.websocket` - WebSocket 协议文档 `STREAMING-IMPLEMENTATION-NOTES.txt` - 流式传输实现说明 ### 外部库 `websockets` - Python WebSocket 库 (https://websockets.readthedocs.io/) `requests` - Python HTTP 库 (现有)