crewai 扫描项目 形成向量数据库
要使用 CrewAI 扫描本地项目代码或文档并将其转化为向量数据库(Vector Database),通常需要结合 CrewAI 的 Tools(工具系统) 以及 LangChain 或 LlamaIndex 等数据处理框架。
由于 CrewAI 本身是一个 Agent(智能体)编排框架,它并不直接存储数据,而是通过给 Agent 赋予“读取文件”和“写入/查询向量库”的能力来完成这个任务。
以下是实现该功能的完整架构和核心代码实现:
🛠️ 核心开发步骤
1. 环境准备与依赖安装
你需要安装 CrewAI 以及处理向量库的依赖(这里以常用的 Chroma 库和 OpenAI 嵌入模型为例):
bash
pip install crewai crewai-tools langchain-openai langchain-community chromadb tqdm
请谨慎使用此类代码。
2. 完整代码实现
你可以创建一个 ingest_project.py 脚本,运行以下代码将项目目录扫描并存入向量库:
python
import os
from pathlib import Path
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import GenericLoader
from langchain_community.document_loaders.parsers import LanguageParser
from langchain.text_splitter import LanguageSplitter
# 1. 设置 API 密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
# 2. 定义自定义工具:让 Agent 可以扫描项目并构建向量库
@tool("Project Scanner and Vector DB Creator")
def scan_project_to_vector_db(project_path: str, db_persist_path: str) -> str:
"""
扫描指定路径的项目代码和文档,将其切片并生成向量,最后保存到 Chroma 向量数据库中。
:param project_path: 项目根目录的绝对或相对路径
:param db_persist_path: 向量数据库保存的本地路径
"""
try:
path = Path(project_path)
if not path.exists():
return f"错误:路径 {project_path} 不存在。"
# 使用 LangChain 的通用加载器解析代码(以 Python 为例,可根据项目调整)
print(f"正在扫描项目: {project_path} ...")
loader = GenericLoader.from_filesystem(
path,
glob="**/*",
suffixes=[".py", ".md", ".txt"], # 扫描 Python 文件和文档
parser=LanguageParser(language="python", parser_threshold=500)
)
documents = loader.load()
if not documents:
return "没有找到有效的代码或文档文件。"
# 针对代码/文本进行智能切片
splitter = LanguageSplitter.from_language(language="python", chunk_size=2000, chunk_overlap=200)
texts = splitter.split_documents(documents)
print(f"成功将项目切分为 {len(texts)} 个文本块。")
# 初始化嵌入模型并写入向量数据库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
print("正在生成向量并写入数据库...")
vector_db = Chroma.from_documents(
documents=texts,
embedding=embeddings,
persist_directory=db_persist_path
)
return f"成功!项目已扫描并存入向量数据库,保存在:{db_persist_path}"
except Exception as e:
return f"执行过程中发生错误: {str(e)}"
# 3. 定义负责处理该任务的 Agent
db_architect = Agent(
role="数据架构师与向量库专家",
goal="负责高效扫描指定项目的源代码和技术文档,并将其结构化转化为高效的向量数据库。",
backstory="你是一名精通代码分析和检索增强生成(RAG)系统的专家。你的职责是确保所有的代码逻辑和文档能被准确地切片并索引到向量数据库中,以便后续的高级搜索。",
tools=[scan_project_to_vector_db],
verbose=True,
memory=False
)
# 4. 定义扫描任务
scan_task = Task(
description=(
"执行项目扫描任务。你需要使用工具将位于 '{project_dir}' 的项目文件进行扫描,"
"并将生成的向量数据库持久化保存到 '{vector_db_dir}' 目录中。"
),
expected_output="关于向量数据库构建成功与否的状态报告,包含切片数量或保存路径。",
agent=db_architect
)
# 5. 组装 Crew 并运行
project_crew = Crew(
agents=[db_architect],
tasks=[scan_task],
process=Process.sequential
)
# 6. 启动任务(填入你的项目路径和想保存数据库的路径)
if __name__ == "__main__":
result = project_crew.kickoff(inputs={
"project_dir": "./my_target_project", # 目标扫描项目路径
"vector_db_dir": "./project_vector_db" # 向量数据库输出路径
})
print("\n--- 最终执行结果 ---")
print(result)
请谨慎使用此类代码。
💡 进阶:如何让 CrewAI 随后使用这个向量库?
一旦上面的脚本运行完毕,你的项目就已经变成了一个本地的向量数据库(Chroma DB)。接下来,你可以使用 CrewAI 自带的 ChromaSearchTool,让其他 Agent 去查询这个数据库来回答关于项目的问题:
python
from crewai_tools import ChromaSearchTool
# 加载刚刚生成的向量数据库工具
project_search_tool = ChromaSearchTool(
chroma_folder_path="./project_vector_db" # 指向刚刚生成的目录
)
# 定义一个代码审查或问答 Agent
code_analyst = Agent(
role="资深代码审查员",
goal="使用向量数据库回答关于项目架构和代码细节的问题。",
backstory="你对这个项目的整套代码了如指掌,能够快速定位 Bug 并解释系统架构。",
tools=[project_search_tool], # 将向量库查询工具赋予它
verbose=True
)
请谨慎使用此类代码。
📋 最佳实践建议
文件过滤(Ignore):在实际项目中,请务必在加载器中排除
.git、__pycache__、node_modules以及虚拟环境(如.venv)目录,否则会产生大量无效向量并消耗大量 Token。切片策略:代码文件和普通文本的切片逻辑不同。代码切片(如
LanguageSplitter)会尽量保持函数和类的完整性,避免生硬截断导致大模型无法理解代码逻辑。
请问你希望扫描的项目主要使用什么编程语言(如 Python, Java, Go)?另外,你打算使用哪家供应商的 Embedding 嵌入模型(如 OpenAI, HuggingFace 本地模型, Ollama)?明确这些信息我可以为你调整更精准的代码切片和免 API 消耗的本地化配置。
为了对 PHP 项目进行精准扫描并构建向量数据库,我们需要调整两点核心配置:
代码切片器(Splitter):使用专门针对 PHP 语法的切片器,确保类(Class)、函数(Function)和命名空间(Namespace)不会被硬生生截断。
文件过滤器:自动识别
.php后缀,并排除类似vendor/(Composer 依赖包)、var/cache/等无用目录。
以下是专门针对 PHP 项目 优化的 CrewAI 向量库构建方案:
🛠️ 1. 环境准备
确保安装了处理 PHP 代码和向量库所需的依赖:
bash
pip install crewai crewai-tools langchain-openai langchain-community chromadb tree-sitter tree-sitter-php
请谨慎使用此类代码。
(注:tree-sitter-php 用于让 LangChain 能够完美解析 PHP 的语法树结构)
📂 2. PHP 项目专属扫描脚本
你可以创建 ingest_php_project.py,并将以下代码粘贴进去运行:
python
import os
from pathlib import Path
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import GenericLoader
from langchain_community.document_loaders.parsers import LanguageParser
from langchain.text_splitter import LanguageSplitter
# 1. 配置您的 API 密钥
os.environ["OPENAI_API_KEY"] = "your-openai-api-key"
# 2. 定义 PHP 专属项目扫描工具
@tool("PHP Project Scanner and Vector DB Creator")
def scan_php_project_to_vector_db(project_path: str, db_persist_path: str) -> str:
"""
专门扫描 PHP 项目源代码(.php)及配置文件、文档,切片并生成向量存入 Chroma 数据库。
自动忽略 vendor 等第三方依赖目录。
"""
try:
path = Path(project_path)
if not path.exists():
return f"错误:项目路径 {project_path} 不存在。"
print(f"🔍 正在扫描 PHP 项目: {project_path} ...")
# 使用通用加载器解析 PHP 代码
loader = GenericLoader.from_filesystem(
path,
glob="**/*",
suffixes=[".php", ".md", ".json"], # 扫描 php 源码、文档和 composer.json
exclude=[
"**/vendor/**", # 排除 Composer 依赖包
"**/var/cache/**", # 排除 Symfony/Laravel 等缓存
"**/storage/**", # 排除日志和上传文件
"**/.git/**" # 排除版本控制目录
],
parser=LanguageParser(language="php", parser_threshold=200)
)
documents = loader.load()
if not documents:
return "未找到任何有效的 PHP 文件或文档。"
# 针对 PHP 语法进行智能切片(保持类和方法的完整性)
# chunk_size 可根据项目复杂度调整,1500-2000 适合保持 PHP 类的上下文
splitter = LanguageSplitter.from_language(language="php", chunk_size=1500, chunk_overlap=150)
texts = splitter.split_documents(documents)
print(f"📊 成功将 PHP 项目切分为 {len(texts)} 个代码/文本块。")
# 初始化向量模型并写入数据库
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
print("🚀 正在生成向量并写入 Chroma 数据库...")
vector_db = Chroma.from_documents(
documents=texts,
embedding=embeddings,
persist_directory=db_persist_path
)
return f"🎉 成功!PHP 项目已转换为向量库。保存路径:{db_persist_path}"
except Exception as e:
return f"❌ 执行失败,错误信息: {str(e)}" # 3. 组装 CrewAI Agent 与 Task php_architect = Agent( role="PHP 架构专家", goal="对 PHP 项目进行全盘扫描、分析,提取出结构清晰的代码片段并建立向量索引。", backstory="你是一位拥有 10 年经验的 PHP 资深架构师,精通 Composer 生态、Laravel/Symfony 等框架。你能准确区分业务核心代码与第三方垃圾文件。", tools=[scan_php_project_to_vector_db], verbose=True ) scan_task = Task( description="扫描位于 '{project_dir}' 的 PHP 项目,排除第三方依赖,将核心代码转换为向量库并存至 '{vector_db_dir}'。", expected_output="包含扫描文件数、切片数量以及数据库保存路径的成功报告。", agent=php_architect ) php_project_crew = Crew( agents=[php_architect], tasks=[scan_task], process=Process.sequential ) # 4. 运行 if __name__ == "__main__": result = php_project_crew.kickoff(inputs={ "project_dir": "./my_php_project", # 替换为你的 PHP 项目路径 "vector_db_dir": "./php_project_vdb" # 向量库输出路径 }) print("\n--- 执行结果 ---") print(result)
请谨慎使用此类代码。
💡 3. 后续如何让 Agent 理解你的 PHP 项目?
数据库建立好后,你可以利用生成的向量库搭建一个 “PHP 代码 AI 助手”。比如让它帮你重构代码、寻找业务漏洞或解释复杂的 Controller 逻辑:
python
from crewai import Agent, Task, Crew
from crewai_tools import ChromaSearchTool
# 1. 载入刚刚生成的 PHP 向量工具
php_code_search_tool = ChromaSearchTool(
chroma_folder_path="./php_project_vdb"
)
# 2. 定义分析智能体
php_analyst = Agent(
role="资深 PHP 审计员",
goal="通过查询项目的向量数据库,快速定位业务逻辑、回答代码设计问题、查找潜在 Bug。",
backstory="你对当前 PHP 项目的代码细节了如指掌,能够快速根据需求找出对应的 Controller、Model 或 Service 类。",
tools=[php_code_search_tool],
verbose=True
)
# 3. 提问任务
audit_task = Task(
description="分析该项目中用户登录(Login)或者鉴权(Authentication)的逻辑是在哪个文件实现的?请详细描述其流程。",
expected_output="一份关于登录逻辑的文件路径清单和核心步骤解析。",
agent=php_analyst
)
# 运行审计
Crew(agents=[php_analyst], tasks=[audit_task]).kickoff()
请谨慎使用此类代码。
为了让这个 PHP 扫描器更契合你的实际场景,你可以告诉我:
你的项目使用了哪个 PHP 框架?(例如:(A) Laravel / (B) Symfony / (C) ThinkPHP / (D) 原生 PHP)
你是否需要将 Embedding 模型替换为 完全免费的本地模型(如通过 Ollama 运行本地模型,免去 OpenAI API 密钥)?