(
request: Request,
data: ChunkQueryRequest,
collection_id: str,
auth_info: Dict = Depends(auth_info_required),
)
| 19 | response_model=BaseListResponse, |
| 20 | ) |
| 21 | async def api_query_chunk( |
| 22 | request: Request, |
| 23 | data: ChunkQueryRequest, |
| 24 | collection_id: str, |
| 25 | auth_info: Dict = Depends(auth_info_required), |
| 26 | ): |
| 27 | chunks: List[Chunk] = await query_chunks( |
| 28 | collection_ids=[collection_id], |
| 29 | top_k=data.top_k, |
| 30 | max_tokens=data.max_tokens, |
| 31 | query_text=data.query_text, |
| 32 | score_threshold=data.score_threshold, |
| 33 | ) |
| 34 | return BaseListResponse( |
| 35 | data=[chunk.to_response_dict() for chunk in chunks], |
| 36 | fetched_count=len(chunks), |
| 37 | ).model_dump(exclude_none=True) |
| 38 | |
| 39 | |
| 40 | @router.get( |
nothing calls this directly
no test coverage detected