✨ feat(database_optimizer): 添加数据库优化工程师智能体功能

- 使用 DeepSeek 模型进行智能分析
- 集成 PostgreSQL MCP 服务器工具
- 加载数据库知识库
- 提供数据库优化建议

数据库优化工程师.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`作为模型(相当于拥有一个聪明的大脑,总是能够做出正确的决策)
+@dataclass
-    - [ ] 精通关系型数据库系统的理论知识以及`postgresql`的实现细节(知识库)
+class DatabaseOptimizerConfig:
-    - [ ] 能够使用`postgresql mcp server`来分析现有数据库设计(行为)
+    """数据库优化器配置"""
+    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"
 
-    """
+class DatabaseOptimizer:
-    # 然后在notebook中加载
+    """PostgreSQL 数据库优化引擎"""
-    from dotenv import load_dotenv
+    
+    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文件
+    def _load_environment(self):
-    load_dotenv()
+        """加载环境变量"""
-    from agno.agent import Agent, RunResponse  # noqa
+        load_dotenv()
-    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
 
-    # Create a knowledge base of PDFs from URLs
+    def _setup_knowledge_base(self) -> CombinedKnowledgeBase:
-    # pdf_url_kb = PDFUrlKnowledgeBase(
+        """设置知识库"""
-    #     urls=["https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf"],
+        local_pdf_kb = PDFKnowledgeBase(
-    #     # Use LanceDB as the vector database and store embeddings in the `recipes` table
+            path=self.config.pdf_path,
-    #     vector_db=LanceDb(
+            vector_db=LanceDb(
-    #         table_name="recipes",
+                table_name="database_fundamentals",
-    #         uri="tmp/lancedb",
+                uri=self.config.vector_db_path,
-    #         search_type=SearchType.vector,
+                search_type=SearchType.vector,
-    #         embedder=OpenAIEmbedder(id="text-embedding-3-small"),
+                embedder=OpenAIEmbedder(id="text-embedding-3-small"),
-    #     ),
+            ),
-    # )
+        )
-    # Create Local PDF knowledge base
+        
-    local_pdf_kb = PDFKnowledgeBase(
+        knowledge_base = CombinedKnowledgeBase(
-        path="D:\\Sources\\DONGJAK-TOOLS\\pdfs\\Database Fundamentals.pdf",
+            sources=[local_pdf_kb],
-        vector_db=LanceDb(
+            vector_db=LanceDb(
-            table_name="database_fundamentals",
+                table_name="combined_documents",
-            uri="tmp/lancedb",
+                uri=self.config.vector_db_path,
-            search_type=SearchType.vector,
+                search_type=SearchType.vector,
-            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
+                embedder=OpenAIEmbedder(id="text-embedding-3-small"),
-        ),
+            ),
-    )
+        )
+        knowledge_base.load()
+        return knowledge_base
 
-    # Combine knowledge bases
+    def _setup_postgres_tools(self) -> MCPTools:
-    knowledge_base = CombinedKnowledgeBase(
+        """设置 PostgreSQL 工具"""
-        sources=[
+        server_params = StdioServerParameters(
-            local_pdf_kb,
+            command="cmd",
-        ],
+            args=[
-        vector_db=LanceDb(
+                "/c",
-            table_name="combined_documents",
+                "npx",
-            uri="tmp/lancedb",
+                "-y",
-            search_type=SearchType.vector,
+                "@modelcontextprotocol/server-postgres",
-            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
+                self.config.db_connection,
-        ),
+            ],
-    )
+            env={},
-    # Load the knowledge base: Comment after first run as the knowledge base is already loaded
+        )
-    knowledge_base.load()
+        return MCPTools(server_params=server_params)
 
-    server_params = StdioServerParameters(
+    def _create_agent(self) -> Agent:
-        command="cmd",  # 或 "uvx"，取决于你的安装方式
+        """创建智能体"""
-        args=[
+        return Agent(
-            "/c",
+            model=DeepSeek(id=self.config.model_id),
-            "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"),
             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
+    def analyze_database(self, query: str, stream: bool = True):
-        # run: RunResponse = agent.run("Share a 2 sentence horror story")
+        """分析数据库"""
-        # print(run.content)
+        with self.postgres_tools:
+            self.agent.print_response(query, stream=stream)
 
-        # Print the response in the terminal
+def main():
-        agent.print_response("看下aq这个数据库", stream=True)
+    """主入口函数"""
+    optimizer = DatabaseOptimizer()
+    optimizer.analyze_database("看下aq这个数据库")
+
+if __name__ == "__main__":
+    main()