CrewAI智能体开发：Qdrant 向量搜索工具

用于 CrewAI 智能体的 Qdrant 向量搜索工具的语义搜索功能。

王国平

645人浏览 · 2025-12-31 07:00:23

王国平 · 2025-12-31 07:00:23 发布

概述

Qdrant 向量搜索工具通过利用向量相似性搜索引擎 Qdrant，为您的 CrewAI 智能体提供语义搜索功能。此工具允许您的智能体使用语义相似性在 Qdrant 集合中存储的文档中进行搜索。

安装安装所需的软件包

uv add qdrant-client

基本用法这是一个如何使用该工具的最小示例

from crewai import Agent
from crewai_tools import QdrantVectorSearchTool, QdrantConfig

# Initialize the tool with QdrantConfig
qdrant_tool = QdrantVectorSearchTool(
    qdrant_config=QdrantConfig(
        qdrant_url="your_qdrant_url",
        qdrant_api_key="your_qdrant_api_key",
        collection_name="your_collection"
    )
)

# Create an agent that uses the tool
agent = Agent(
    role="Research Assistant",
    goal="Find relevant information in documents",
    tools=[qdrant_tool]
)

# The tool will automatically use OpenAI embeddings
# and return the 3 most relevant results with scores > 0.35

完整的工作示例这是一个完整的示例，展示了如何

从 PDF 中提取文本
使用 OpenAI 生成嵌入
存储在 Qdrant 中
为语义搜索创建 CrewAI 智能体 RAG 工作流

import os
import uuid
import pdfplumber
from openai import OpenAI
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process, LLM
from crewai_tools import QdrantVectorSearchTool
from qdrant_client import QdrantClient
from qdrant_client.models import PointStruct, Distance, VectorParams

# Load environment variables
load_dotenv()

# Initialize OpenAI client
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# Extract text from PDF
def extract_text_from_pdf(pdf_path):
    text = []
    with pdfplumber.open(pdf_path) as pdf:
        for page in pdf.pages:
            page_text = page.extract_text()
            if page_text:
                text.append(page_text.strip())
    return text

# Generate OpenAI embeddings
def get_openai_embedding(text):
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-large"
    )
    return response.data[0].embedding

# Store text and embeddings in Qdrant
def load_pdf_to_qdrant(pdf_path, qdrant, collection_name):
    # Extract text from PDF
    text_chunks = extract_text_from_pdf(pdf_path)

    # Create Qdrant collection
    if qdrant.collection_exists(collection_name):
        qdrant.delete_collection(collection_name)
    qdrant.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(size=3072, distance=Distance.COSINE)
    )

    # Store embeddings
    points = []
    for chunk in text_chunks:
        embedding = get_openai_embedding(chunk)
        points.append(PointStruct(
            id=str(uuid.uuid4()),
            vector=embedding,
            payload={"text": chunk}
        ))
    qdrant.upsert(collection_name=collection_name, points=points)

# Initialize Qdrant client and load data
qdrant = QdrantClient(
    url=os.getenv("QDRANT_URL"),
    api_key=os.getenv("QDRANT_API_KEY")
)
collection_name = "example_collection"
pdf_path = "path/to/your/document.pdf"
load_pdf_to_qdrant(pdf_path, qdrant, collection_name)

# Initialize Qdrant search tool
from crewai_tools import QdrantConfig

qdrant_tool = QdrantVectorSearchTool(
    qdrant_config=QdrantConfig(
        qdrant_url=os.getenv("QDRANT_URL"),
        qdrant_api_key=os.getenv("QDRANT_API_KEY"),
        collection_name=collection_name,
        limit=3,
        score_threshold=0.35
    )
)

# Create CrewAI agents
search_agent = Agent(
    role="Senior Semantic Search Agent",
    goal="Find and analyze documents based on semantic search",
    backstory="""You are an expert research assistant who can find relevant
    information using semantic search in a Qdrant database.""",
    tools=[qdrant_tool],
    verbose=True
)

answer_agent = Agent(
    role="Senior Answer Assistant",
    goal="Generate answers to questions based on the context provided",
    backstory="""You are an expert answer assistant who can generate
    answers to questions based on the context provided.""",
    tools=[qdrant_tool],
    verbose=True
)

# Define tasks
search_task = Task(
    description="""Search for relevant documents about the {query}.
    Your final answer should include:
    - The relevant information found
    - The similarity scores of the results
    - The metadata of the relevant documents""",
    agent=search_agent
)

answer_task = Task(
    description="""Given the context and metadata of relevant documents,
    generate a final answer based on the context.""",
    agent=answer_agent
)

# Run CrewAI workflow
crew = Crew(
    agents=[search_agent, answer_agent],
    tasks=[search_task, answer_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff(
    inputs={"query": "What is the role of X in the document?"}
)
print(result)

工具参数

必填参数

qdrant_config (QdrantConfig)：包含所有 Qdrant 设置的配置对象

QdrantConfig 参数

qdrant_url (str)：您的 Qdrant 服务器的 URL
qdrant_api_key (str, 可选)：用于 Qdrant 身份验证的 API 密钥
collection_name (str)：要搜索的 Qdrant 集合的名称
limit (int)：要返回的最大结果数（默认值：3）
score_threshold (float)：最小相似性分数阈值（默认值：0.35）
filter (Any, 可选)：用于高级过滤的 Qdrant Filter 实例（默认值：None）

可选工具参数

custom_embedding_fn (Callable[[str], list[float]])：用于文本向量化的自定义函数
qdrant_package (str)：Qdrant 的基本包路径（默认值：“qdrant_client”）
client (Any)：预初始化的 Qdrant 客户端（可选）

高级过滤QdrantVectorSearchTool 支持强大的过滤功能来优化您的搜索结果

动态过滤在搜索中使用 filter_by 和 filter_value 参数以动态过滤结果

# Agent will use these parameters when calling the tool
# The tool schema accepts filter_by and filter_value
# Example: search with category filter
# Results will be filtered where category == "technology"

使用 QdrantConfig 的预设过滤器对于复杂的过滤，请在您的配置中使用 Qdrant Filter 实例

from qdrant_client.http import models as qmodels
from crewai_tools import QdrantVectorSearchTool, QdrantConfig

# Create a filter for specific conditions
preset_filter = qmodels.Filter(
    must=[
        qmodels.FieldCondition(
            key="category",
            match=qmodels.MatchValue(value="research")
        ),
        qmodels.FieldCondition(
            key="year",
            match=qmodels.MatchValue(value=2024)
        )
    ]
)

# Initialize tool with preset filter
qdrant_tool = QdrantVectorSearchTool(
    qdrant_config=QdrantConfig(
        qdrant_url="your_url",
        qdrant_api_key="your_key",
        collection_name="your_collection",
        filter=preset_filter  # Preset filter applied to all searches
    )
)

组合过滤器该工具自动将来自 QdrantConfig 的预设过滤器与来自 filter_by 和 filter_value 的动态过滤器组合在一起

# If QdrantConfig has a preset filter for category="research"
# And the search uses filter_by="year", filter_value=2024
# Both filters will be combined (AND logic)

搜索参数该工具在其架构中接受这些参数

query (str)：用于查找相似文档的搜索查询
filter_by (str, 可选)：要过滤的元数据字段
filter_value (Any, 可选)：要过滤的值

返回格式该工具以 JSON 格式返回结果

[
  {
    "metadata": {
      // Any metadata stored with the document
    },
    "context": "The actual text content of the document",
    "distance": 0.95  // Similarity score
  }
]

默认嵌入默认情况下，该工具使用 OpenAI 的 text-embedding-3-large 模型进行向量化。这需要

在环境中设置 OpenAI API 密钥：OPENAI_API_KEY

自定义嵌入在以下情况下，您可能希望使用自己的嵌入函数而不是默认嵌入模型

想要使用不同的嵌入模型（例如，Cohere、HuggingFace、Ollama 模型）
需要通过使用开源嵌入模型来降低成本
对向量维度或嵌入质量有特定要求
想要使用特定领域的嵌入（例如，用于医学或法律文本）

这是一个使用 HuggingFace 模型的示例

from transformers import AutoTokenizer, AutoModel
import torch

# Load model and tokenizer
tokenizer = AutoTokenizer.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')
model = AutoModel.from_pretrained('sentence-transformers/all-MiniLM-L6-v2')

def custom_embeddings(text: str) -> list[float]:
    # Tokenize and get model outputs
    inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
    outputs = model(**inputs)

    # Use mean pooling to get text embedding
    embeddings = outputs.last_hidden_state.mean(dim=1)

    # Convert to list of floats and return
    return embeddings[0].tolist()

# Use custom embeddings with the tool
from crewai_tools import QdrantConfig

tool = QdrantVectorSearchTool(
    qdrant_config=QdrantConfig(
        qdrant_url="your_url",
        qdrant_api_key="your_key",
        collection_name="your_collection"
    ),
    custom_embedding_fn=custom_embeddings  # Pass your custom function
)

错误处理该工具处理这些特定错误

如果未安装 qdrant-client，则引发 ImportError（可选择自动安装）
如果未设置 QDRANT_URL，则引发 ValueError
如果缺少 qdrant-client，则提示使用 uv add qdrant-client 安装

环境变量所需环境变量

export QDRANT_URL="your_qdrant_url"  # If not provided in constructor
export QDRANT_API_KEY="your_api_key"  # If not provided in constructor
export OPENAI_API_KEY="your_openai_key"  # If using default embeddings

《动手学PyTorch建模与应用:从深度学习到大模型》是一本从零基础上手深度学习和大模型的PyTorch实战指南。全书共11章，前6章涵盖深度学习基础，包括张量运算、神经网络原理、数据预处理及卷积神经网络等；后5章进阶探讨图像、文本、音频建模技术，并结合Transformer架构解析大语言模型的开发实践。书中通过房价预测、图像分类等案例讲解模型构建方法，每章附有动手练习题，帮助读者巩固实战能力。内容兼顾数学原理与工程实现，适配PyTorch框架最新技术发展趋势。