MemOS-MCP

""" MemOS官方API文档生成器 自动生成API文档、测试用例和使用示例 """ import json import yaml from pathlib import Path from typing import Dict, Any, List from datetime import datetime from official_api_server import create_official_api_app, setup_exception_handlers class APIDocumentationGenerator: """API文档生成器""" def __init__(self, output_dir: str = "./api_docs"): self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) # 创建API应用实例 self.app, self.mvp_manager = create_official_api_app() setup_exception_handlers(self.app) def generate_openapi_spec(self) -> Dict[str, Any]: """生成OpenAPI规范""" return self.app.openapi() def save_openapi_spec(self, format: str = "json") -> str: """保存OpenAPI规范""" spec = self.generate_openapi_spec() if format.lower() == "json": file_path = self.output_dir / "openapi.json" with open(file_path, 'w', encoding='utf-8') as f: json.dump(spec, f, indent=2, ensure_ascii=False) elif format.lower() == "yaml": file_path = self.output_dir / "openapi.yaml" with open(file_path, 'w', encoding='utf-8') as f: yaml.dump(spec, f, default_flow_style=False, allow_unicode=True) else: raise ValueError("格式必须是 'json' 或 'yaml'") return str(file_path) def generate_api_reference(self) -> str: """生成API参考文档""" spec = self.generate_openapi_spec() markdown_content = f"""# MemOS官方API参考文档 **版本**: {spec.get('info', {}).get('version', '1.0.0')} **生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ## 概述 {spec.get('info', {}).get('description', 'MemOS官方API')} ## 基础信息 - **基础URL**: `http://localhost:8000` - **认证方式**: 无需认证（个人使用场景） - **响应格式**: JSON - **字符编码**: UTF-8 ## 标准响应格式 ### 成功响应 ```json {{ "success": true, "message": "操作成功", "data": {{}}, "timestamp": "2025-07-15T12:00:00", "request_id": "req_1234567890" }} ``` ### 错误响应 ```json {{ "success": false, "error_code": "ERROR_CODE", "error_message": "错误描述", "details": {{}}, "timestamp": "2025-07-15T12:00:00" }} ``` ## API端点 """ # 遍历所有路径和方法 paths = spec.get('paths', {}) for path, methods in paths.items(): for method, details in methods.items(): if method.upper() in ['GET', 'POST', 'PUT', 'DELETE', 'PATCH']: markdown_content += self._format_endpoint_doc(path, method.upper(), details) # 保存文档 doc_path = self.output_dir / "api_reference.md" with open(doc_path, 'w', encoding='utf-8') as f: f.write(markdown_content) return str(doc_path) def _format_endpoint_doc(self, path: str, method: str, details: Dict[str, Any]) -> str: """格式化单个端点的文档""" doc = f""" ### {method} {path} **摘要**: {details.get('summary', '无描述')} **描述**: {details.get('description', '无详细描述')} """ # 请求参数 if 'requestBody' in details: doc += "**请求体**:\n```json\n" # 这里可以添加请求体示例 doc += "{\n // 请求参数\n}\n```\n\n" # 响应示例 responses = details.get('responses', {}) if '200' in responses: doc += "**成功响应** (200):\n```json\n" doc += "{\n \"success\": true,\n \"message\": \"操作成功\",\n \"data\": {}\n}\n```\n\n" doc += "---\n" return doc def generate_usage_examples(self) -> str: """生成使用示例""" examples = """# MemOS API使用示例 ## Python示例 ### 基础设置 ```python import requests import json # API基础URL BASE_URL = "http://localhost:8000" # 通用请求头 HEADERS = { "Content-Type": "application/json", "Accept": "application/json" } ``` ### 1. 健康检查 ```python def check_health(): response = requests.get(f"{BASE_URL}/health") return response.json() # 使用示例 health_status = check_health() print(f"系统状态: {health_status['data']['api_status']}") ``` ### 2. 创建记忆 ```python def create_memory(content, tags=None): data = { "content": content, "tags": tags or [], "metadata": {} } response = requests.post( f"{BASE_URL}/memories", headers=HEADERS, json=data ) return response.json() # 使用示例 result = create_memory( content="这是一个测试记忆", tags=["测试", "API"] ) print(f"创建结果: {result['message']}") ``` ### 3. 搜索记忆 ```python def search_memories(query, top_k=5): data = { "query": query, "top_k": top_k, "use_reranker": True, "use_feedback_boost": True } response = requests.post( f"{BASE_URL}/memories/search", headers=HEADERS, json=data ) return response.json() # 使用示例 results = search_memories("测试记忆") print(f"找到 {results['data']['total_count']} 条记忆") for result in results['data']['results']: print(f"- {result['memory']['content'][:50]}...") ``` ### 4. 提供反馈 ```python def provide_feedback(memory_id, feedback_type): data = { "memory_id": memory_id, "feedback_type": feedback_type } response = requests.post( f"{BASE_URL}/memories/feedback", headers=HEADERS, json=data ) return response.json() # 使用示例 feedback_result = provide_feedback("123", "thumbs_up") print(f"反馈结果: {feedback_result['message']}") ``` ### 5. 获取系统状态 ```python def get_system_status(): response = requests.get(f"{BASE_URL}/status") return response.json() # 使用示例 status = get_system_status() print(f"运行模式: {status['data']['mode']}") print(f"嵌入模型: {status['data']['model']}") ``` ### 6. 获取性能指标 ```python def get_performance_metrics(): response = requests.get(f"{BASE_URL}/metrics") return response.json() # 使用示例 metrics = get_performance_metrics() print(f"平均添加时间: {metrics['data']['memory_add_avg_time']:.3f}s") print(f"平均检索时间: {metrics['data']['memory_search_avg_time']:.3f}s") ``` ## JavaScript示例 ### 基础设置 ```javascript const BASE_URL = 'http://localhost:8000'; const apiRequest = async (endpoint, options = {}) => { const response = await fetch(`${BASE_URL}${endpoint}`, { headers: { 'Content-Type': 'application/json', 'Accept': 'application/json', ...options.headers }, ...options }); return response.json(); }; ``` ### 创建记忆 ```javascript const createMemory = async (content, tags = []) => { return await apiRequest('/memories', { method: 'POST', body: JSON.stringify({ content, tags, metadata: {} }) }); }; // 使用示例 createMemory('这是一个JavaScript测试记忆', ['JS', 'API']) .then(result => console.log('创建成功:', result.message)); ``` ### 搜索记忆 ```javascript const searchMemories = async (query, topK = 5) => { return await apiRequest('/memories/search', { method: 'POST', body: JSON.stringify({ query, top_k: topK, use_reranker: true, use_feedback_boost: true }) }); }; // 使用示例 searchMemories('JavaScript测试') .then(results => { console.log(`找到 ${results.data.total_count} 条记忆`); results.data.results.forEach(result => { console.log(`- ${result.memory.content.substring(0, 50)}...`); }); }); ``` ## cURL示例 ### 健康检查 ```bash curl -X GET "http://localhost:8000/health" \\ -H "Accept: application/json" ``` ### 创建记忆 ```bash curl -X POST "http://localhost:8000/memories" \\ -H "Content-Type: application/json" \\ -d '{ "content": "这是一个cURL测试记忆", "tags": ["curl", "test"], "metadata": {} }' ``` ### 搜索记忆 ```bash curl -X POST "http://localhost:8000/memories/search" \\ -H "Content-Type: application/json" \\ -d '{ "query": "测试记忆", "top_k": 5, "use_reranker": true, "use_feedback_boost": true }' ``` ## 错误处理 ### 常见错误代码 - `HTTP_400`: 请求参数错误 - `HTTP_404`: 资源不存在 - `HTTP_500`: 内部服务器错误 - `HTTP_503`: 服务不可用 ### 错误处理示例 ```python try: response = requests.post(f"{BASE_URL}/memories", json=data) response.raise_for_status() # 抛出HTTP错误 result = response.json() if not result.get('success'): print(f"API错误: {result.get('error_message')}") else: print(f"操作成功: {result.get('message')}") except requests.exceptions.RequestException as e: print(f"请求错误: {e}") except json.JSONDecodeError as e: print(f"JSON解析错误: {e}") ``` """ # 保存示例文档 examples_path = self.output_dir / "usage_examples.md" with open(examples_path, 'w', encoding='utf-8') as f: f.write(examples) return str(examples_path) def generate_all_docs(self) -> Dict[str, str]: """生成所有文档""" docs = {} print("📖 生成API文档...") # 生成OpenAPI规范 docs['openapi_json'] = self.save_openapi_spec('json') docs['openapi_yaml'] = self.save_openapi_spec('yaml') print(f"✅ OpenAPI规范已生成") # 生成API参考文档 docs['api_reference'] = self.generate_api_reference() print(f"✅ API参考文档已生成") # 生成使用示例 docs['usage_examples'] = self.generate_usage_examples() print(f"✅ 使用示例已生成") # 生成文档索引 index_content = f"""# MemOS官方API文档 **生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ## 文档列表 1. **[API参考文档](api_reference.md)** - 完整的API端点文档 2. **[使用示例](usage_examples.md)** - Python、JavaScript、cURL使用示例 3. **[OpenAPI规范 (JSON)](openapi.json)** - 机器可读的API规范 4. **[OpenAPI规范 (YAML)](openapi.yaml)** - 人类可读的API规范 ## 快速开始 1. 启动API服务器: ```bash python official_api_server.py ``` 2. 访问API文档: http://localhost:8000/docs 3. 测试健康检查: ```bash curl http://localhost:8000/health ``` ## 主要功能 - ✅ 记忆创建和搜索 - ✅ 智能反馈机制 - ✅ 容量管理和监控 - ✅ 性能指标统计 - ✅ 系统状态查询 - ✅ 标准化错误处理 - ✅ 完整的API文档 ## 技术特性 - 基于FastAPI框架 - 标准化响应格式 - 完整的错误处理 - 性能监控中间件 - CORS支持 - 自动API文档生成 """ index_path = self.output_dir / "README.md" with open(index_path, 'w', encoding='utf-8') as f: f.write(index_content) docs['index'] = str(index_path) print(f"✅ 文档索引已生成") print(f"\n📁 所有文档已保存到: {self.output_dir}") return docs if __name__ == "__main__": # 生成API文档 generator = APIDocumentationGenerator() docs = generator.generate_all_docs() print("\n📋 生成的文档:") for doc_type, path in docs.items(): print(f" {doc_type}: {path}") print("\n🚀 文档生成完成！")