agent_management/agent_templates/docs/POSTGRESQL_AGENT_EXAMPLES.md

PostgreSQL Agent 请求调用示例

服务信息

• 服务名称: PostgreSQL AI Agent
• 版本: 1.0.0
• 框架: LangChain + OpenAI
• 默认端口: 8080

概述

PostgreSQL AI Agent 使用 LangChain 和自然语言处理技术，允许用户使用自然语言查询 PostgreSQL 数据库。

---

API 端点

1. 健康检查

端点: GET /health

请求示例 (curl):

curl http://localhost:8080/health

响应示例:

{
  "status": "healthy",
  "pod_name": "postgresql-agent",
  "template_type": "postgresql_agent",
  "database_connected": true,
  "database_info": "localhost:5432/mydb"
}

---

2. 服务信息

端点: GET /

请求示例 (curl):

curl http://localhost:8080/

响应示例:

{
  "name": "PostgreSQL AI Agent",
  "version": "1.0.0",
  "database": "localhost:5432/postgres",
  "endpoints": {
    "health": "/health",
    "query": "/query"
  }
}

---

3. 自然语言查询

端点: POST /query

使用自然语言查询 PostgreSQL 数据库。

请求体:

{
  "query": "显示所有用户",
  "openai_api_key": "sk-xxx",
  "user_id": "user123",
  "model": "gpt-3.5-turbo"
}

请求示例 (curl):

curl -X POST http://localhost:8080/query \
  -H "Content-Type: application/json" \
  -d '{
    "query": "数据库中有多少个表?",
    "openai_api_key": "sk-xxx",
    "user_id": "user123"
  }'

请求示例 (Python):

import requests

query_data = {
    "query": "显示 users 表中年龄大于25的所有用户",
    "openai_api_key": "sk-xxx",
    "user_id": "user123",
    "model": "gpt-3.5-turbo"
}

response = requests.post(
    "http://localhost:8080/query",
    json=query_data
)
result = response.json()

print(f"Query: {result['query']}")
print(f"Result: {result['result']}")
print(f"Success: {result['success']}")
print(f"Timestamp: {result['timestamp']}")

响应示例:

{
  "query": "数据库中有多少个表?",
  "result": "数据库中有 12 个表",
  "success": true,
  "timestamp": "2026-01-15T10:30:00.000Z"
}

---

查询示例

基础查询:

queries = [
    "显示所有表",
    "列出所有schema",
    "显示 users 表的结构",
    "users 表有多少条记录?",
    "显示最近创建的10条记录"
]

PostgreSQL 特定功能:

queries = [
    "显示所有视图",
    "列出所有索引",
    "显示表的大小",
    "查看数据库的大小",
    "显示所有触发器",
    "列出所有存储过程",
    "显示表的统计信息"
]

统计查询:

queries = [
    "统计每个部门的员工数量",
    "计算订单的总金额",
    "显示每月的销售额",
    "找出销量最高的产品",
    "计算用户的平均年龄"
]

条件查询:

queries = [
    "显示状态为活跃的用户",
    "查找创建时间在最近一周的订单",
    "列出价格高于1000的产品",
    "显示评分大于4.5的商品",
    "查找北京地区的所有客户"
]

关联查询:

queries = [
    "显示每个用户的订单数量",
    "列出有订单的用户",
    "显示每个类别的产品数量",
    "查找购买了特定产品的用户",
    "统计每个城市的订单总额"
]

JSON 查询 (PostgreSQL 特性):

queries = [
    "从 users 表的 metadata JSON 字段中提取 age",
    "查找 metadata 包含特定键的记录",
    "统计 JSON 数组的长度"
]

---

完整使用示例

Python 客户端:

import requests
from typing import Optional, Dict, Any, List

class PostgreSQLAgentClient:
    """PostgreSQL Agent 客户端"""
    
    def __init__(self, base_url: str, openai_api_key: str):
        self.base_url = base_url.rstrip('/')
        self.openai_api_key = openai_api_key
    
    def health_check(self) -> Dict[str, Any]:
        """健康检查"""
        response = requests.get(f"{self.base_url}/health")
        return response.json()
    
    def get_info(self) -> Dict[str, Any]:
        """获取服务信息"""
        response = requests.get(f"{self.base_url}/")
        return response.json()
    
    def query(
        self,
        query: str,
        user_id: Optional[str] = None,
        model: str = "gpt-3.5-turbo"
    ) -> Dict[str, Any]:
        """执行自然语言查询"""
        data = {
            "query": query,
            "openai_api_key": self.openai_api_key,
            "user_id": user_id,
            "model": model
        }
        response = requests.post(f"{self.base_url}/query", json=data)
        return response.json()
    
    def batch_query(self, queries: List[str], user_id: Optional[str] = None) -> List[Dict]:
        """批量查询"""
        results = []
        for q in queries:
            result = self.query(q, user_id)
            results.append(result)
        return results


# 使用示例
client = PostgreSQLAgentClient(
    base_url="http://localhost:8080",
    openai_api_key="sk-xxx"
)

# 1. 健康检查
health = client.health_check()
print(f"Status: {health['status']}")
print(f"Database: {health['database_info']}")
print(f"Connected: {health['database_connected']}\n")

# 2. 单个查询
result = client.query("显示所有表")
print(f"Query: {result['query']}")
print(f"Result: {result['result']}\n")

# 3. 批量查询
queries = [
    "数据库中有多少个表?",
    "users 表有多少条记录?",
    "显示 users 表的前5条记录",
    "统计每个城市的用户数量"
]

print("Batch Queries:")
results = client.batch_query(queries, user_id="user123")
for i, result in enumerate(results, 1):
    print(f"\n{i}. {result['query']}")
    if result['success']:
        print(f"   {result['result']}")
    else:
        print(f"   Error: Failed to execute query")

---

数据库监控工具:

import requests
import time
from datetime import datetime
from rich.console import Console
from rich.table import Table

class PostgreSQLMonitor:
    """PostgreSQL 数据库监控工具"""
    
    def __init__(self, base_url: str, openai_api_key: str):
        self.base_url = base_url
        self.openai_api_key = openai_api_key
        self.console = Console()
    
    def query(self, query: str) -> str:
        """执行查询"""
        response = requests.post(
            f"{self.base_url}/query",
            json={
                "query": query,
                "openai_api_key": self.openai_api_key
            }
        )
        result = response.json()
        return result.get('result', '') if result.get('success') else 'N/A'
    
    def get_database_stats(self) -> Dict:
        """获取数据库统计信息"""
        stats = {
            "database_size": self.query("数据库的大小是多少?"),
            "table_count": self.query("有多少个表?"),
            "connection_count": self.query("当前有多少个数据库连接?"),
            "cache_hit_ratio": self.query("缓存命中率是多少?")
        }
        return stats
    
    def display_stats(self, stats: Dict):
        """显示统计信息"""
        table = Table(title="PostgreSQL Database Statistics")
        table.add_column("Metric", style="cyan")
        table.add_column("Value", style="green")
        
        for metric, value in stats.items():
            table.add_row(metric.replace('_', ' ').title(), str(value))
        
        self.console.print(table)
    
    def monitor(self, interval: int = 60):
        """持续监控"""
        self.console.print("[bold blue]PostgreSQL Monitor Started[/bold blue]")
        self.console.print(f"Refresh interval: {interval} seconds\n")
        
        try:
            while True:
                self.console.clear()
                self.console.print(f"[yellow]Last Update: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}[/yellow]\n")
                
                stats = self.get_database_stats()
                self.display_stats(stats)
                
                time.sleep(interval)
        except KeyboardInterrupt:
            self.console.print("\n[yellow]Monitoring stopped[/yellow]")


# 使用示例
monitor = PostgreSQLMonitor(
    base_url="http://localhost:8080",
    openai_api_key="sk-xxx"
)

# 获取一次统计信息
stats = monitor.get_database_stats()
monitor.display_stats(stats)

# 或者持续监控 (每60秒刷新)
# monitor.monitor(interval=60)

---

数据迁移辅助工具:

import requests
from typing import List, Dict

class PostgreSQLMigrationHelper:
    """PostgreSQL 数据迁移辅助工具"""
    
    def __init__(self, base_url: str, openai_api_key: str):
        self.base_url = base_url
        self.openai_api_key = openai_api_key
    
    def query(self, query: str) -> str:
        """执行查询"""
        response = requests.post(
            f"{self.base_url}/query",
            json={
                "query": query,
                "openai_api_key": self.openai_api_key
            }
        )
        result = response.json()
        return result.get('result', '') if result.get('success') else ''
    
    def get_table_schema(self, table_name: str) -> str:
        """获取表结构"""
        return self.query(f"显示 {table_name} 表的详细结构")
    
    def get_all_tables(self) -> str:
        """获取所有表名"""
        return self.query("列出所有表名")
    
    def get_table_constraints(self, table_name: str) -> str:
        """获取表约束"""
        return self.query(f"显示 {table_name} 表的所有约束")
    
    def get_table_indexes(self, table_name: str) -> str:
        """获取表索引"""
        return self.query(f"显示 {table_name} 表的所有索引")
    
    def get_foreign_keys(self, table_name: str) -> str:
        """获取外键关系"""
        return self.query(f"显示 {table_name} 表的外键关系")
    
    def generate_migration_report(self, table_name: str) -> Dict:
        """生成迁移报告"""
        return {
            "table": table_name,
            "schema": self.get_table_schema(table_name),
            "constraints": self.get_table_constraints(table_name),
            "indexes": self.get_table_indexes(table_name),
            "foreign_keys": self.get_foreign_keys(table_name)
        }


# 使用示例
helper = PostgreSQLMigrationHelper(
    base_url="http://localhost:8080",
    openai_api_key="sk-xxx"
)

# 获取所有表
tables = helper.get_all_tables()
print(f"All Tables:\n{tables}\n")

# 生成特定表的迁移报告
table_name = "users"
report = helper.generate_migration_report(table_name)

print(f"Migration Report for '{table_name}':")
print(f"\nSchema:\n{report['schema']}")
print(f"\nConstraints:\n{report['constraints']}")
print(f"\nIndexes:\n{report['indexes']}")
print(f"\nForeign Keys:\n{report['foreign_keys']}")

---

性能分析工具:

import requests
from typing import List, Dict
import pandas as pd

class PostgreSQLPerformanceAnalyzer:
    """PostgreSQL 性能分析工具"""
    
    def __init__(self, base_url: str, openai_api_key: str):
        self.base_url = base_url
        self.openai_api_key = openai_api_key
    
    def query(self, query: str) -> str:
        """执行查询"""
        response = requests.post(
            f"{self.base_url}/query",
            json={
                "query": query,
                "openai_api_key": self.openai_api_key
            }
        )
        result = response.json()
        return result.get('result', '') if result.get('success') else ''
    
    def get_slow_queries(self) -> str:
        """获取慢查询"""
        return self.query("显示最慢的10个查询")
    
    def get_table_sizes(self) -> str:
        """获取表大小"""
        return self.query("显示所有表的大小，按大小降序排列")
    
    def get_index_usage(self) -> str:
        """获取索引使用情况"""
        return self.query("显示索引使用统计")
    
    def get_cache_stats(self) -> str:
        """获取缓存统计"""
        return self.query("显示缓存命中率统计")
    
    def get_connection_stats(self) -> str:
        """获取连接统计"""
        return self.query("显示数据库连接统计信息")
    
    def analyze_table(self, table_name: str) -> str:
        """分析表性能"""
        return self.query(f"分析 {table_name} 表的性能")


# 使用示例
analyzer = PostgreSQLPerformanceAnalyzer(
    base_url="http://localhost:8080",
    openai_api_key="sk-xxx"
)

print("=== Performance Analysis ===\n")

# 1. 慢查询
print("Slow Queries:")
print(analyzer.get_slow_queries())
print()

# 2. 表大小
print("Table Sizes:")
print(analyzer.get_table_sizes())
print()

# 3. 索引使用
print("Index Usage:")
print(analyzer.get_index_usage())
print()

# 4. 缓存统计
print("Cache Statistics:")
print(analyzer.get_cache_stats())
print()

# 5. 分析特定表
print("Analyze 'users' table:")
print(analyzer.analyze_table("users"))

---

环境变量配置

# 服务配置
export SERVICE_HOST="0.0.0.0"
export SERVICE_PORT="8080"
export POD_NAME="postgresql-agent"
export TEMPLATE_TYPE="postgresql_agent"

# PostgreSQL 数据库配置
export POSTGRES_HOST="localhost"
export POSTGRES_PORT="5432"
export POSTGRES_USER="postgres"
export POSTGRES_PASSWORD="your-password"
export POSTGRES_DATABASE="mydb"

# 启动服务
python postgresql_agent.py

---

Docker Compose 示例

version: '3.8'

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: testdb
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
  
  postgresql-agent:
    build:
      context: .
      dockerfile: postgresql_agent.Dockerfile
    environment:
      POSTGRES_HOST: postgres
      POSTGRES_PORT: 5432
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DATABASE: testdb
      SERVICE_PORT: 8080
    ports:
      - "8080:8080"
    depends_on:
      - postgres

volumes:
  postgres_data:

---

PostgreSQL 特性支持

Agent 支持 PostgreSQL 的特殊功能:

• ✅ JSON/JSONB 查询
• ✅ 数组类型
• ✅ 全文搜索
• ✅ 窗口函数
• ✅ CTEs (Common Table Expressions)
• ✅ 视图和物化视图
• ✅ 触发器和存储过程
• ✅ 分区表

---

注意事项

1. API Key: OpenAI API key 从请求传入，确保安全
2. 数据库连接: 需要正确配置数据库连接参数
3. 权限控制: 建议使用只读用户进行查询
4. 查询限制: 设置合理的查询超时和结果限制
5. 错误处理: 检查 success 字段确认查询是否成功
6. PostgreSQL 版本: 支持 PostgreSQL 12+
7. 成本控制: 监控 OpenAI API 使用量
8. 模型选择: gpt-4 更准确但成本更高