Airweave - 让 AI 代理搜索任何应用的统一知识平台

   git clone https://github.com/airweave-ai/airweave.git   cd airweave

   pip install pre-commit   pre-commit install

   ./start.sh

   cp .env.example .env   # 编辑.env文件配置必要的环境变量

@router.post("/api-keys", response_model=schemas.APIKey)async def create_api_key(    api_key_in: schemas.APIKeyCreate = Body({}),    auth_context: AuthContext = Depends(deps.get_auth_context)):    """创建新的API密钥"""    api_key_obj = await crud.api_key.create(        db=db,         obj_in=api_key_in,         auth_context=auth_context    )    return api_key_obj

@router.get("/collections/search")async def search_collection(    query: str,    collection_id: UUID,    db: AsyncSession = Depends(deps.get_db)):    """在指定集合中执行语义搜索"""    results = await search_service.search(        query=query,        collection_id=collection_id    )    return results

class Auth0ManagementClient:    def __init__(self):        self.domain = settings.AUTH0_DOMAIN        self.client_id = settings.AUTH0_M2M_CLIENT_ID        self.client_secret = settings.AUTH0_M2M_CLIENT_SECRET        self.audience = f"https://{self.domain}/api/v2/"
    async def _get_management_token(self):        """获取Auth0管理API访问令牌"""        async with httpx.AsyncClient() as client:            response = await client.post(                f"https://{self.domain}/oauth/token",                json={                    "client_id": self.client_id,                    "client_secret": self.client_secret,                    "audience": self.audience,                    "grant_type": "client_credentials"                }            )            return response.json()["access_token"]

class QdrantDestination(VectorDBDestination):    """Qdrant向量数据库实现"""        @classmethod    async def create(cls, collection_id: UUID):        """创建Qdrant目标实例"""        instance = cls()        instance.collection_id = collection_id        instance.collection_name = str(collection_id)        instance.client = AsyncQdrantClient(            url=settings.QDRANT_URL,            api_key=settings.QDRANT_API_KEY        )        return instance
    async def bulk_insert(self, entities: list[ChunkEntity]):        """批量插入实体到Qdrant"""        points = [            rest.PointStruct(                id=str(entity.id),                vector=entity.embedding,                payload=entity.to_dict()            ) for entity in entities        ]        await self.client.upsert(            collection_name=self.collection_name,            points=points        )