viking_client.py 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263
  1. """
  2. Viking 知识库上传工具模块
  3. 职责:
  4. 1. 将本地临时文件上传到火山引擎 TOS 对象存储
  5. 2. 调用 Viking /api/knowledge/doc/v2/add 接口将 TOS 文件导入知识库
  6. """
  7. import os
  8. import uuid
  9. import asyncio
  10. from concurrent.futures import ThreadPoolExecutor
  11. import httpx
  12. from ..config.config import Config
  13. cfg = Config()
  14. # 线程池用于执行 TOS 同步 SDK 调用
  15. _executor = ThreadPoolExecutor(max_workers=4)
  16. def _search_knowledge_sync(resource_id: str, query: str, limit: int = 5) -> list:
  17. """
  18. 同步检索 Viking 知识库,返回 PointInfo 列表。
  19. 在线程池中调用,避免阻塞事件循环。
  20. 配置:开启重排序(base-multilingual-rerank)、邻近切片扩散1、文本聚合。
  21. """
  22. from vikingdb.knowledge import VikingKnowledge
  23. from vikingdb.auth import APIKey
  24. from vikingdb.knowledge.models.search import SearchKnowledgeRequest
  25. viking_client = VikingKnowledge(
  26. host=cfg.VIKING_HOST,
  27. region="cn-beijing",
  28. auth=APIKey(api_key=cfg.VIKING_API_KEY),
  29. scheme="https",
  30. )
  31. collection = viking_client.collection(resource_id=resource_id)
  32. resp = collection.search_knowledge(SearchKnowledgeRequest(
  33. query=query,
  34. limit=limit,
  35. post_processing={
  36. "rerank_switch": True,
  37. "retrieve_count": max(limit * 3, 15), # 进入重排的候选数,须 >= limit
  38. "chunk_diffusion_count": 1, # 命中切片上下各扩散1片
  39. "chunk_group": True, # 按文档原始顺序重排,适合叙述性文档
  40. },
  41. ))
  42. if resp and resp.result and resp.result.result_list:
  43. return resp.result.result_list
  44. return []
  45. async def search_knowledge(resource_id: str, query: str, limit: int = 5) -> list:
  46. """
  47. 异步检索 Viking 知识库,返回 PointInfo 列表。
  48. result_list 非空表示知识库有命中,为空则降级到联网搜索。
  49. """
  50. loop = asyncio.get_event_loop()
  51. return await loop.run_in_executor(_executor, _search_knowledge_sync, resource_id, query, limit)