From f28025852751cb5942e351a532a46cc813b09c97 Mon Sep 17 00:00:00 2001 From: cruld Date: Mon, 31 Mar 2025 01:00:28 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(database=5Foptimizer):=20?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=95=B0=E6=8D=AE=E5=BA=93=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E5=B7=A5=E7=A8=8B=E5=B8=88=E6=99=BA=E8=83=BD=E4=BD=93=E5=8A=9F?= =?UTF-8?q?=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用 DeepSeek 模型进行智能分析 - 集成 PostgreSQL MCP 服务器工具 - 加载数据库知识库 - 提供数据库优化建议 --- 数据库优化工程师.py | 175 ++++++++++++++++++++++++-------------------- 1 file changed, 94 insertions(+), 81 deletions(-) diff --git a/数据库优化工程师.py b/数据库优化工程师.py index faafe7e..df05f01 100644 --- a/数据库优化工程师.py +++ b/数据库优化工程师.py @@ -1,96 +1,109 @@ -if __name__ == "__main__": - """ - ## 背景 +#!/usr/bin/env python3 +""" +PostgreSQL 数据库优化工程师智能体 - 我有一个`postgresql`数据库,现在需要一个专业的数据库优化工程师,来帮我看下设计是否合理,以及如何优化. +功能: +- 使用 DeepSeek 模型进行智能分析 +- 集成 PostgreSQL MCP 服务器工具 +- 加载数据库知识库 +- 提供数据库优化建议 +""" - ## 目标 +import os +from typing import Optional +from dataclasses import dataclass +from dotenv import load_dotenv - 这个数据库优化工程师智能体应该具备以下能力: +from agno.agent import Agent +from agno.models.deepseek import DeepSeek +from agno.knowledge.pdf import PDFKnowledgeBase +from agno.vectordb.lancedb import LanceDb, SearchType +from agno.embedder.openai import OpenAIEmbedder +from agno.knowledge.combined import CombinedKnowledgeBase +from agno.tools.mcp import MCPTools +from mcp import StdioServerParameters - - [ ] 使用`claude3.7`作为模型(相当于拥有一个聪明的大脑,总是能够做出正确的决策) - - [ ] 精通关系型数据库系统的理论知识以及`postgresql`的实现细节(知识库) - - [ ] 能够使用`postgresql mcp server`来分析现有数据库设计(行为) +@dataclass +class DatabaseOptimizerConfig: + """数据库优化器配置""" + pdf_path: str = "D:\\Sources\\DONGJAK-TOOLS\\pdfs\\Database Fundamentals.pdf" + db_connection: str = "postgresql://postgres:postgres@192.168.1.7:5432/postgres" + model_id: str = "deepseek-chat" + vector_db_path: str = "tmp/lancedb" - """ - # 然后在notebook中加载 - from dotenv import load_dotenv +class DatabaseOptimizer: + """PostgreSQL 数据库优化引擎""" + + def __init__(self, config: Optional[DatabaseOptimizerConfig] = None): + self.config = config or DatabaseOptimizerConfig() + self._load_environment() + self.knowledge_base = self._setup_knowledge_base() + self.postgres_tools = self._setup_postgres_tools() + self.agent = self._create_agent() - # 加载当前目录下的.env文件 - load_dotenv() - from agno.agent import Agent, RunResponse # noqa - from agno.models.deepseek import DeepSeek - from agno.knowledge.pdf_url import PDFUrlKnowledgeBase - from agno.knowledge.pdf import PDFKnowledgeBase - from agno.vectordb.lancedb import LanceDb, SearchType - from agno.embedder.openai import OpenAIEmbedder - from agno.knowledge.combined import CombinedKnowledgeBase - from agno.tools.mcp import MCPTools - from mcp import StdioServerParameters + def _load_environment(self): + """加载环境变量""" + load_dotenv() - # Create a knowledge base of PDFs from URLs - # pdf_url_kb = PDFUrlKnowledgeBase( - # urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"], - # # Use LanceDB as the vector database and store embeddings in the `recipes` table - # vector_db=LanceDb( - # table_name="recipes", - # uri="tmp/lancedb", - # search_type=SearchType.vector, - # embedder=OpenAIEmbedder(id="text-embedding-3-small"), - # ), - # ) - # Create Local PDF knowledge base - local_pdf_kb = PDFKnowledgeBase( - path="D:\\Sources\\DONGJAK-TOOLS\\pdfs\\Database Fundamentals.pdf", - vector_db=LanceDb( - table_name="database_fundamentals", - uri="tmp/lancedb", - search_type=SearchType.vector, - embedder=OpenAIEmbedder(id="text-embedding-3-small"), - ), - ) + def _setup_knowledge_base(self) -> CombinedKnowledgeBase: + """设置知识库""" + local_pdf_kb = PDFKnowledgeBase( + path=self.config.pdf_path, + vector_db=LanceDb( + table_name="database_fundamentals", + uri=self.config.vector_db_path, + search_type=SearchType.vector, + embedder=OpenAIEmbedder(id="text-embedding-3-small"), + ), + ) + + knowledge_base = CombinedKnowledgeBase( + sources=[local_pdf_kb], + vector_db=LanceDb( + table_name="combined_documents", + uri=self.config.vector_db_path, + search_type=SearchType.vector, + embedder=OpenAIEmbedder(id="text-embedding-3-small"), + ), + ) + knowledge_base.load() + return knowledge_base - # Combine knowledge bases - knowledge_base = CombinedKnowledgeBase( - sources=[ - local_pdf_kb, - ], - vector_db=LanceDb( - table_name="combined_documents", - uri="tmp/lancedb", - search_type=SearchType.vector, - embedder=OpenAIEmbedder(id="text-embedding-3-small"), - ), - ) - # Load the knowledge base: Comment after first run as the knowledge base is already loaded - knowledge_base.load() + def _setup_postgres_tools(self) -> MCPTools: + """设置 PostgreSQL 工具""" + server_params = StdioServerParameters( + command="cmd", + args=[ + "/c", + "npx", + "-y", + "@modelcontextprotocol/server-postgres", + self.config.db_connection, + ], + env={}, + ) + return MCPTools(server_params=server_params) - server_params = StdioServerParameters( - command="cmd", # 或 "uvx"，取决于你的安装方式 - args=[ - "/c", - "npx", - "-y", - "@modelcontextprotocol/server-postgres", - "postgresql://postgres:postgres@192.168.1.7:5432/postgres", - ], - env={}, # 可选的环境变量 - ) - with MCPTools(server_params=server_params) as postgres_server: - # 使用mcp_tools - - agent = Agent( - model=DeepSeek(id="deepseek-chat"), + def _create_agent(self) -> Agent: + """创建智能体""" + return Agent( + model=DeepSeek(id=self.config.model_id), markdown=True, - knowledge=knowledge_base, + knowledge=self.knowledge_base, search_knowledge=True, show_tool_calls=True, - tools=[postgres_server], + tools=[self.postgres_tools], ) - # Get the response in a variable - # run: RunResponse = agent.run("Share a 2 sentence horror story") - # print(run.content) + def analyze_database(self, query: str, stream: bool = True): + """分析数据库""" + with self.postgres_tools: + self.agent.print_response(query, stream=stream) - # Print the response in the terminal - agent.print_response("看下aq这个数据库", stream=True) +def main(): + """主入口函数""" + optimizer = DatabaseOptimizer() + optimizer.analyze_database("看下aq这个数据库") + +if __name__ == "__main__": + main()