Generate embeddings for knowledge base files. Args: files: List of dicts, e.g. [{"path": "/outputs/kb_data/...", "description": "..."}] api_url: Custom Embedding API URL. api_key: Custom API Key. model_name: Custom Model Name. multimodal_model: C
(
files: List[Dict[str, Optional[str]]] = Body(..., embed=True),
api_url: Optional[str] = Body(None, embed=True),
api_key: Optional[str] = Body(None, embed=True),
model_name: Optional[str] = Body(None, embed=True),
multimodal_model: Optional[str] = Body(settings.KB_EMBEDDING_MODEL, embed=True),
image_model: Optional[str] = Body(None, embed=True),
video_model: Optional[str] = Body(None, embed=True),
notebook_id: Optional[str] = Body(None, embed=True),
user_id: Optional[str] = Body(None, embed=True),
user: AuthUser = Depends(get_current_user),
)
| 89 | |
| 90 | @router.post("/embedding") |
| 91 | async def create_embedding( |
| 92 | files: List[Dict[str, Optional[str]]] = Body(..., embed=True), |
| 93 | api_url: Optional[str] = Body(None, embed=True), |
| 94 | api_key: Optional[str] = Body(None, embed=True), |
| 95 | model_name: Optional[str] = Body(None, embed=True), |
| 96 | multimodal_model: Optional[str] = Body(settings.KB_EMBEDDING_MODEL, embed=True), |
| 97 | image_model: Optional[str] = Body(None, embed=True), |
| 98 | video_model: Optional[str] = Body(None, embed=True), |
| 99 | notebook_id: Optional[str] = Body(None, embed=True), |
| 100 | user_id: Optional[str] = Body(None, embed=True), |
| 101 | user: AuthUser = Depends(get_current_user), |
| 102 | ): |
| 103 | """ |
| 104 | Generate embeddings for knowledge base files. |
| 105 | |
| 106 | Args: |
| 107 | files: List of dicts, e.g. [{"path": "/outputs/kb_data/...", "description": "..."}] |
| 108 | api_url: Custom Embedding API URL. |
| 109 | api_key: Custom API Key. |
| 110 | model_name: Custom Model Name. |
| 111 | multimodal_model: Custom Multimodal Model Name (default: gemini-2.5-flash). |
| 112 | image_model: Custom Image Model Name. |
| 113 | video_model: Custom Video Model Name. |
| 114 | """ |
| 115 | try: |
| 116 | process_list = [] |
| 117 | user_email = _canonical_user_email(user) |
| 118 | resolved_user_id = _canonical_user_id(user) |
| 119 | |
| 120 | for f in files: |
| 121 | web_path = f.get("path") |
| 122 | desc = f.get("description") |
| 123 | |
| 124 | if not web_path: |
| 125 | continue |
| 126 | |
| 127 | local_path = _resolve_user_owned_output_path(web_path, user) |
| 128 | if local_path.exists(): |
| 129 | process_list.append({ |
| 130 | "path": str(local_path), |
| 131 | "description": desc |
| 132 | }) |
| 133 | else: |
| 134 | log.warning(f"File not found locally: {local_path}") |
| 135 | |
| 136 | if not process_list: |
| 137 | return { |
| 138 | "success": False, |
| 139 | "message": "No valid files found to process." |
| 140 | } |
| 141 | |
| 142 | # Define vector store location |
| 143 | if notebook_id and user_email: |
| 144 | from fastapi_app.routers.kb import _vector_store_dir |
| 145 | vector_store_dir = _vector_store_dir(user_email, notebook_id, resolved_user_id) |
| 146 | elif user_email: |
| 147 | vector_store_dir = get_outputs_root() / "kb_data" / user_email / "vector_store" |
| 148 | else: |
nothing calls this directly
no test coverage detected