requirements.txt:
fastapi==0.104.1
uvicorn[standard]==0.24.0
pydantic==2.5.0
python-dotenv==1.0.0
python-multipart==0.0.6
sqlalchemy==2.0.23
alembic==1.12.1
psycopg2-binary==2.9.9
redis==5.0.1
httpx==0.25.2
aiofiles==23.2.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
prometheus-client==0.19.0
opentelemetry-api==1.20.0
opentelemetry-sdk==1.20.0
opentelemetry-instrumentation-fastapi==0.41b0
pytest==7.4.3
pytest-asyncio==0.21.1
pytest-cov==4.1.0

import os
from typing import Optional
from pydantic_settings import BaseSettings
from dotenv import load_dotenv

load_dotenv()

class Settings(BaseSettings):
    """應用配置類"""
    
    # 應用配置
    APP_NAME: str = "AI Agent Service"
    VERSION: str = "1.0.0"
    DEBUG: bool = False
    
    # API配置
    API_V1_STR: str = "/api/v1"
    PROJECT_NAME: str = "ai-agent-service"
    
    # 安全配置
    SECRET_KEY: str = os.getenv("SECRET_KEY", "your-secret-key-here")
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    
    # 數(shù)據(jù)庫配置
    DATABASE_URL: str = os.getenv("DATABASE_URL", "postgresql://user:pass@localhost/ai_agent")
    REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379")
    
    # LLM配置
    OPENAI_API_KEY: Optional[str] = os.getenv("OPENAI_API_KEY")
    OPENAI_BASE_URL: Optional[str] = os.getenv("OPENAI_BASE_URL")
    MODEL_NAME: str = os.getenv("MODEL_NAME", "gpt-3.5-turbo")
    
    # 速率限制
    RATE_LIMIT_PER_MINUTE: int = 60
    
    class Config:
        case_sensitive = True
        env_file = ".env"

settings = Settings()

import logging
import sys
from loguru import logger
from app.core.config import settings

def setup_logging():
    """配置結構化日志"""
    
    # 移除默認處理器
    logger.remove()
    
    # 添加控制臺處理器
    logger.add(
        sys.stdout,
        level="DEBUG" if settings.DEBUG else "INFO",
        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
        backtrace=True,
        diagnose=settings.DEBUG
    )
    
    # 添加文件處理器（生產(chǎn)環(huán)境）
    if not settings.DEBUG:
        logger.add(
            "logs/ai_agent.log",
            rotation="10 MB",
            retention="30 days",
            level="INFO",
            format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}"
        )
    
    return logger

# 創(chuàng)建全局logger實例
log = setup_logging()

from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
from datetime import datetime
from enum import Enum

class MessageRole(str, Enum):
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"
    TOOL = "tool"

class ToolType(str, Enum):
    FUNCTION = "function"
    API = "api"
    DATABASE = "database"

class ChatRequest(BaseModel):
    """聊天請求模型"""
    message: str = Field(..., description="用戶消息")
    conversation_id: Optional[str] = Field(None, description="會話ID")
    tools: Optional[List[str]] = Field(None, description="要使用的工具列表")
    stream: bool = Field(False, description="是否流式響應")
    
    class Config:
        json_schema_extra = {
            "example": {
                "message": "你好，請幫我查詢天氣",
                "conversation_id": "conv_123",
                "tools": ["weather", "calculator"],
                "stream": False
            }
        }

class ChatResponse(BaseModel):
    """聊天響應模型"""
    message: str = Field(..., description="AI回復")
    conversation_id: str = Field(..., description="會話ID")
    tool_calls: Optional[List[Dict[str, Any]]] = Field(None, description="工具調(diào)用")
    usage: Optional[Dict[str, int]] = Field(None, description="使用統(tǒng)計")
    
    class Config:
        json_schema_extra = {
            "example": {
                "message": "你好！我是AI助手，很高興為您服務。",
                "conversation_id": "conv_123",
                "tool_calls": None,
                "usage": {"prompt_tokens": 10, "completion_tokens": 20}
            }
        }

class ToolCall(BaseModel):
    """工具調(diào)用模型"""
    tool_name: str = Field(..., description="工具名稱")
    parameters: Dict[str, Any] = Field(..., description="工具參數(shù)")
    
    class Config:
        json_schema_extra = {
            "example": {
                "tool_name": "weather",
                "parameters": {"city": "北京"}
            }
        }

class ToolResponse(BaseModel):
    """工具響應模型"""
    tool_name: str = Field(..., description="工具名稱")
    result: Any = Field(..., description="工具執(zhí)行結果")
    success: bool = Field(..., description="執(zhí)行是否成功")
    error_message: Optional[str] = Field(None, description="錯誤信息")

class Conversation(BaseModel):
    """會話模型"""
    id: str = Field(..., description="會話ID")
    user_id: str = Field(..., description="用戶ID")
    title: str = Field(..., description="會話標題")
    created_at: datetime = Field(..., description="創(chuàng)建時間")
    updated_at: datetime = Field(..., description="更新時間")
    message_count: int = Field(..., description="消息數(shù)量")

class HealthCheck(BaseModel):
    """健康檢查響應"""
    status: str = Field(..., description="服務狀態(tài)")
    version: str = Field(..., description="服務版本")
    timestamp: datetime = Field(..., description="檢查時間")

class ErrorResponse(BaseModel):
    """錯誤響應模型"""
    error: str = Field(..., description="錯誤類型")
    message: str = Field(..., description="錯誤信息")
    details: Optional[Dict[str, Any]] = Field(None, description="錯誤詳情")

from fastapi import FastAPI, Depends, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from contextlib import asynccontextmanager
import time

from app.core.config import settings
from app.core.logging import log
from app.api.endpoints import chat, tools, admin
from app.api.dependencies import get_current_user
from app.models.schemas import HealthCheck

@asynccontextmanager
async def lifespan(app: FastAPI):
    """應用生命周期管理"""
    # 啟動時執(zhí)行
    log.info("?? Starting AI Agent Service")
    yield
    # 關閉時執(zhí)行
    log.info("?? Shutting down AI Agent Service")

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    description="企業(yè)級AI Agent微服務",
    docs_url="/docs" if settings.DEBUG else None,
    redoc_url="/redoc" if settings.DEBUG else None,
    lifespan=lifespan
)

# 中間件配置
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"] if settings.DEBUG else ["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["*"] if settings.DEBUG else ["yourdomain.com"]
)

@app.middleware("http")
async def add_process_time_header(request, call_next):
    """添加請求處理時間頭"""
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

# 健康檢查端點
@app.get("/health", response_model=HealthCheck, tags=["health"])
async def health_check():
    return HealthCheck(
        status="healthy",
        version=settings.VERSION,
        timestamp=time.time()
    )

# 注冊路由
app.include_router(
    chat.router,
    prefix=settings.API_V1_STR,
    tags=["chat"],
    dependencies=[Depends(get_current_user)]
)

app.include_router(
    tools.router,
    prefix=settings.API_V1_STR,
    tags=["tools"],
    dependencies=[Depends(get_current_user)]
)

app.include_router(
    admin.router,
    prefix=settings.API_V1_STR + "/admin",
    tags=["admin"],
    dependencies=[Depends(get_current_user)]
)

@app.get("/")
async def root():
    return {"message": f"Welcome to {settings.APP_NAME}"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "app.main:app",
        host="0.0.0.0",
        port=8000,
        reload=settings.DEBUG,
        log_level="info" if settings.DEBUG else "warning"
    )

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
from typing import Optional

from app.services.chat_service import ChatService
from app.models.schemas import ChatRequest, ChatResponse, ErrorResponse
from app.api.dependencies import get_chat_service
from app.core.logging import log

router = APIRouter()

@router.post(
    "/chat",
    response_model=ChatResponse,
    responses={
        400: {"model": ErrorResponse},
        429: {"model": ErrorResponse},
        500: {"model": ErrorResponse}
    }
)
async def chat_endpoint(
    request: ChatRequest,
    chat_service: ChatService = Depends(get_chat_service)
):
    """處理聊天請求"""
    try:
        log.info(f"Processing chat request for conversation: {request.conversation_id}")
        
        if request.stream:
            # 流式響應處理
            return StreamingResponse(
                chat_service.stream_chat(request),
                media_type="text/event-stream"
            )
        else:
            # 普通響應
            return await chat_service.process_chat(request)
            
    except Exception as e:
        log.error(f"Chat processing error: {str(e)}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Chat processing failed: {str(e)}"
        )

@router.get("/conversations/{conversation_id}")
async def get_conversation(conversation_id: str):
    """獲取會話歷史"""
    # 實現(xiàn)獲取會話邏輯
    return {"conversation_id": conversation_id, "messages": []}

@router.delete("/conversations/{conversation_id}")
async def delete_conversation(conversation_id: str):
    """刪除會話"""
    # 實現(xiàn)刪除會話邏輯
    return {"message": f"Conversation {conversation_id} deleted"}

from typing import AsyncGenerator, Optional
from app.models.schemas import ChatRequest, ChatResponse
from app.services.llm_service import LLMService
from app.core.logging import log

class ChatService:
    """聊天 服務"""
    
    def __init__(self, llm_service: LLMService):
        self.llm_service = llm_service
    
    async def process_chat(self, request: ChatRequest) -> ChatResponse:
        """處理聊天請求"""
        try:
            # 處理消息邏輯
            response = await self.llm_service.generate_response(
                message=request.message,
                conversation_id=request.conversation_id,
                tools=request.tools
            )
            
            return ChatResponse(
                message=response["content"],
                conversation_id=response["conversation_id"],
                tool_calls=response.get("tool_calls"),
                usage=response.get("usage")
            )
            
        except Exception as e:
            log.error(f"Chat processing error: {str(e)}")
            raise
    
    async def stream_chat(self, request: ChatRequest) -> AsyncGenerator[str, None]:
        """流式聊天響應"""
        try:
            async for chunk in self.llm_service.stream_response(
                message=request.message,
                conversation_id=request.conversation_id,
                tools=request.tools
            ):
                yield f"data: {chunk}\n\n"
                
        except Exception as e:
            log.error(f"Stream chat error: {str(e)}")
            yield f"data: {{'error': '{str(e)}'}}\n\n"

import httpx
from typing import AsyncGenerator, List, Optional, Dict, Any
from app.core.config import settings
from app.core.logging import log

class LLMService:
    """LLM服務基類"""
    
    async def generate_response(
        self, 
        message: str, 
        conversation_id: Optional[str] = None,
        tools: Optional[List[str]] = None
    ) -> Dict[str, Any]:
        """生成響應"""
        raise NotImplementedError
    
    async def stream_response(
        self, 
        message: str, 
        conversation_id: Optional[str] = None,
        tools: Optional[List[str]] = None
    ) -> AsyncGenerator[str, None]:
        """流式生成響應"""
        raise NotImplementedError

class OpenAIService(LLMService):
    """OpenAI服務實現(xiàn)"""
    
    def __init__(self):
        self.api_key = settings.OPENAI_API_KEY
        self.base_url = settings.OPENAI_BASE_URL or "https://api.openai.com/v1"
        self.model = settings.MODEL_NAME
        self.client = httpx.AsyncClient(
            base_url=self.base_url,
            headers={"Authorization": f"Bearer {self.api_key}"},
            timeout=30.0
        )
    
    async def generate_response(self, message: str, conversation_id: Optional[str] = None, tools: Optional[List[str]] = None) -> Dict[str, Any]:
        try:
            messages = [{"role": "user", "content": message}]
            
            response = await self.client.post(
                "/chat/completions",
                json={
                    "model": self.model,
                    "messages": messages,
                    "max_tokens": 1000,
                    "temperature": 0.7
                }
            )
            
            if response.status_code != 200:
                raise Exception(f"OpenAI API error: {response.text}")
            
            data = response.json()
            return {
                "content": data["choices"][0]["message"]["content"],
                "conversation_id": conversation_id or f"conv_{hash(message)}",
                "usage": data.get("usage")
            }
            
        except Exception as e:
            log.error(f"OpenAI service error: {str(e)}")
            raise
    
    async def stream_response(self, message: str, conversation_id: Optional[str] = None, tools: Optional[List[str]] = None) -> AsyncGenerator[str, None]:
        try:
            messages = [{"role": "user", "content": message}]
            
            async with self.client.stream(
                "POST",
                "/chat/completions",
                json={
                    "model": self.model,
                    "messages": messages,
                    "max_tokens": 1000,
                    "temperature": 0.7,
                    "stream": True
                }
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        if data != "[DONE]":
                            yield data
                            
        except Exception as e:
            log.error(f"OpenAI stream error: {str(e)}")
            yield f'{{"error": "{str(e)}"}}'

from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
from app.core.logging import log

class ToolParameter(BaseModel):
    """工具參數(shù)定義"""
    name: str = Field(..., description="參數(shù)名稱")
    type: str = Field(..., description="參數(shù)類型")
    description: str = Field(..., description="參數(shù)描述")
    required: bool = Field(True, description="是否必需")

class BaseTool(ABC):
    """工具基類"""
    
    def __init__(self):
        self.name = self.__class__.__name__.lower().replace('tool', '')
        self.description = "A general purpose tool"
        self.parameters = []
    
    @abstractmethod
    async def execute(self, **kwargs) -> Any:
        """執(zhí)行工具"""
        pass
    
    def get_schema(self) -> Dict[str, Any]:
        """獲取工具定義"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": [
                param.dict() for param in self.parameters
            ]
        }

class CalculatorTool(BaseTool):
    """計算器工具"""
    
    def __init__(self):
        super().__init__()
        self.name = "calculator"
        self.description = "Perform mathematical calculations"
        self.parameters = [
            ToolParameter(
                name="expression",
                type="string",
                description="Mathematical expression to evaluate",
                required=True
            )
        ]
    
    async def execute(self, expression: str) -> float:
        """執(zhí)行計算"""
        try:
            # 安全評估數(shù)學表達式
            allowed_chars = set('0123456789+-*/.() ')
            if not all(c in allowed_chars for c in expression):
                raise ValueError("Invalid characters in expression")
            
            result = eval(expression)  # 注意：生產(chǎn)環(huán)境需要更安全的方式
            log.info(f"Calculator tool executed: {expression} = {result}")
            return result
            
        except Exception as e:
            log.error(f"Calculator error: {str(e)}")
            raise ValueError(f"Calculation failed: {str(e)}")

from typing import Dict, List, Optional
from app.tools.base import BaseTool
from app.core.logging import log

class ToolRegistry:
    """工具注冊表"""
    
    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}
    
    def register_tool(self, tool: BaseTool):
        """注冊工具"""
        self._tools[tool.name] = tool
        log.info(f"Tool registered: {tool.name}")
    
    def get_tool(self, name: str) -> Optional[BaseTool]:
        """獲取工具"""
        return self._tools.get(name)
    
    def list_tools(self) -> List[Dict]:
        """列出所有工具"""
        return [tool.get_schema() for tool in self._tools.values()]
    
    async def execute_tool(self, name: str, **kwargs) -> Any:
        """執(zhí)行工具"""
        tool = self.get_tool(name)
        if not tool:
            raise ValueError(f"Tool not found: {name}")
        
        return await tool.execute(**kwargs)

# 全局工具注冊表實例
tool_registry = ToolRegistry()

# 注冊默認工具
def register_default_tools():
    """注冊默認工具"""
    from app.tools.base import CalculatorTool
    
    calculator = CalculatorTool()
    tool_registry.register_tool(calculator)

# 初始化時注冊工具
register_default_tools()

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings
from app.core.logging import log

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """驗證密碼"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """生成密碼哈希"""
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
    """創(chuàng)建訪問令牌"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def verify_token(token: str) -> Optional[dict]:
    """驗證令牌"""
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app.core.security import verify_token
from app.core.logging import log
from app.services.llm_service import LLMService, OpenAIService
from app.services.chat_service import ChatService

security = HTTPBearer()

async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    """獲取當前用戶"""
    token = credentials.credentials
    payload = verify_token(token)
    if payload is None:
        log.warning("Invalid authentication token")
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    # 這里可以添加用戶驗證邏輯
    return payload

def get_llm_service() -> LLMService:
    """獲取LLM服務實例"""
    return OpenAIService()

def get_chat_service(llm_service: LLMService = Depends(get_llm_service)) -> ChatService:
    """獲取聊天 服務實例"""
    return ChatService(llm_service)

FROM python:3.11-slim

WORKDIR /app

# 安裝系統(tǒng)依賴
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 復制依賴文件
COPY requirements.txt .

# 安裝Python依賴
RUN pip install --no-cache-dir -r requirements.txt

# 復制應用代碼
COPY . .

# 創(chuàng)建非root用戶
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 啟動命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

version: ‘3.8'

services:
ai-agent:
build: .
ports:
- “8000:8000”
environment:
- DEBUG=true
- DATABASE_URL=postgresql://user:pass@db:5432/ai_agent
- REDIS_URL=redis://redis:6379
depends_on:
- db
- redis
volumes:
- ./logs:/app/logs

db:
image: postgres:13
environment:
- POSTGRES_DB=ai_agent
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
volumes:
- postgres_data:/var/lib/postgresql/data

redis:
image: redis:7-alpine
volumes:
- redis_data:/data

volumes:
postgres_data:
redis_data:

from sqlalchemy import Column, String, Text, DateTime, Integer, Boolean, JSON, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy import create_engine
from datetime import datetime
import uuid
from app.core.config import settings

Base = declarative_base()

def generate_uuid():
    return str(uuid.uuid4())

class User(Base):
    """用戶模型"""
    __tablename__ = "users"
    
    id = Column(String, primary_key=True, default=generate_uuid)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    is_superuser = Column(Boolean, default=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    # 關系
    conversations = relationship("Conversation", back_populates="user")
    api_keys = relationship("APIKey", back_populates="user")

class Conversation(Base):
    """會話模型"""
    __tablename__ = "conversations"
    
    id = Column(String, primary_key=True, default=generate_uuid)
    user_id = Column(String, ForeignKey("users.id"), nullable=False)
    title = Column(String(200), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    message_count = Column(Integer, default=0)
    
    # 關系
    user = relationship("User", back_populates="conversations")
    messages = relationship("Message", back_populates="conversation")

class Message(Base):
    """消息模型"""
    __tablename__ = "messages"
    
    id = Column(String, primary_key=True, default=generate_uuid)
    conversation_id = Column(String, ForeignKey("conversations.id"), nullable=False)
    role = Column(String(20), nullable=False)  # user, assistant, system, tool
    content = Column(Text, nullable=False)
    tool_calls = Column(JSON, nullable=True)
    tool_results = Column(JSON, nullable=True)
    token_count = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)
    
    # 關系
    conversation = relationship("Conversation", back_populates="messages")

class APIKey(Base):
    """API密鑰模型"""
    __tablename__ = "api_keys"
    
    id = Column(String, primary_key=True, default=generate_uuid)
    user_id = Column(String, ForeignKey("users.id"), nullable=False)
    name = Column(String(100), nullable=False)
    key_hash = Column(String(255), nullable=False, unique=True, index=True)
    is_active = Column(Boolean, default=True)
    last_used = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=datetime.utcnow)
    expires_at = Column(DateTime, nullable=True)
    
    # 關系
    user = relationship("User", back_populates="api_keys")

class ToolCallLog(Base):
    """工具調(diào)用日志"""
    __tablename__ = "tool_call_logs"
    
    id = Column(String, primary_key=True, default=generate_uuid)
    user_id = Column(String, ForeignKey("users.id"), nullable=False)
    tool_name = Column(String(100), nullable=False)
    parameters = Column(JSON, nullable=True)
    result = Column(Text, nullable=True)
    success = Column(Boolean, default=True)
    error_message = Column(Text, nullable=True)
    duration_ms = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)

# 數(shù)據(jù)庫引擎和會話
engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db():
    """獲取數(shù)據(jù)庫會話"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def create_tables():
    """創(chuàng)建數(shù)據(jù)庫表"""
    Base.metadata.create_all(bind=engine)

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

from app.models.database import get_db, User, Conversation, ToolCallLog
from app.models.schemas import Conversation as ConversationSchema
from app.api.dependencies import get_current_user
from app.core.logging import log

router = APIRouter()

@router.get("/users/{user_id}/conversations", response_model=List[ConversationSchema])
async def get_user_conversations(
    user_id: str,
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """獲取用戶會話列表"""
    # 檢查權限
    if current_user.get("user_id") != user_id and not current_user.get("is_superuser", False):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Not enough permissions"
        )
    
    conversations = db.query(Conversation).filter(
        Conversation.user_id == user_id
    ).offset(skip).limit(limit).all()
    
    return conversations

@router.get("/metrics/tool-usage")
async def get_tool_usage_metrics(
    days: int = 7,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """獲取工具使用統(tǒng)計"""
    if not current_user.get("is_superuser", False):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    
    # 實現(xiàn)工具使用統(tǒng)計查詢
    return {"message": "Tool usage metrics endpoint"}

@router.delete("/conversations/{conversation_id}")
async def admin_delete_conversation(
    conversation_id: str,
    db: Session = Depends(get_db),
    current_user: dict = Depends(get_current_user)
):
    """管理員刪除會話"""
    if not current_user.get("is_superuser", False):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    
    # 實現(xiàn)管理員刪除會話邏輯
    return {"message": f"Conversation {conversation_id} deleted by admin"}

from prometheus_client import Counter, Histogram, Gauge, generate_latest
from prometheus_client import REGISTRY
from fastapi import Request, Response
import time
from app.core.logging import log

# 定義指標
REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP Requests',
    ['method', 'endpoint', 'status_code']
)

REQUEST_DURATION = Histogram(
    'http_request_duration_seconds',
    'HTTP request duration in seconds',
    ['method', 'endpoint']
)

ACTIVE_REQUESTS = Gauge(
    'http_requests_active',
    'Active HTTP requests'
)

LLM_CALL_COUNT = Counter(
    'llm_calls_total',
    'Total LLM API calls',
    ['provider', 'model', 'status']
)

TOOL_CALL_COUNT = Counter(
    'tool_calls_total',
    'Total tool calls',
    ['tool_name', 'status']
)

class PrometheusMiddleware:
    """Prometheus監(jiān)控中間件"""
    
    def __init__(self, app):
        self.app = app
    
    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        
        start_time = time.time()
        method = scope["method"]
        path = self._get_path(scope["path"])
        
        ACTIVE_REQUESTS.inc()
        
        async def send_wrapper(message):
            if message["type"] == "http.response.start":
                status_code = message["status"]
                REQUEST_COUNT.labels(method=method, endpoint=path, status_code=status_code).inc()
                REQUEST_DURATION.labels(method=method, endpoint=path).observe(time.time() - start_time)
            
            await send(message)
        
        try:
            await self.app(scope, receive, send_wrapper)
        finally:
            ACTIVE_REQUESTS.dec()
    
    def _get_path(self, path: str) -> str:
        """規(guī)范化路徑用于指標"""
        if path.startswith("/api/v1"):
            # 提取API版本后的第一部分作為端點
            parts = path.split("/")
            if len(parts) > 3:
                return f"/api/v1/{parts[3]}"
        return path

def metrics_endpoint(request: Request) -> Response:
    """提供Prometheus指標"""
    return Response(generate_latest(REGISTRY), media_type="text/plain")

def record_llm_call(provider: str, model: str, success: bool = True):
    """記錄LLM調(diào)用"""
    status = "success" if success else "error"
    LLM_CALL_COUNT.labels(provider=provider, model=model, status=status).inc()

def record_tool_call(tool_name: str, success: bool = True):
    """記錄工具調(diào)用"""
    status = "success" if success else "error"
    TOOL_CALL_COUNT.labels(tool_name=tool_name, status=status).inc()

from app.core.monitoring import PrometheusMiddleware, metrics_endpoint

# 添加監(jiān)控中間件（在中間件配置部分）
app.add_middleware(PrometheusMiddleware)

# 添加指標端點（在路由部分）
@app.get("/metrics")
async def metrics():
    from app.core.monitoring import metrics_endpoint
    return metrics_endpoint

import redis
import json
import pickle
from typing import Any, Optional
from app.core.config import settings
from app.core.logging import log

class CacheManager:
    """Redis緩存管理器"""
    
    def __init__(self):
        self.redis_client = redis.from_url(settings.REDIS_URL, decode_responses=False)
        self.default_ttl = 3600  # 1小時默認TTL
    
    async def get(self, key: str) -> Optional[Any]:
        """獲取緩存值"""
        try:
            data = self.redis_client.get(key)
            if data:
                return pickle.loads(data)
            return None
        except Exception as e:
            log.error(f"Cache get error for key {key}: {e}")
            return None
    
    async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> bool:
        """設置緩存值"""
        try:
            serialized_value = pickle.dumps(value)
            expire_time = ttl if ttl is not None else self.default_ttl
            self.redis_client.setex(key, expire_time, serialized_value)
            return True
        except Exception as e:
            log.error(f"Cache set error for key {key}: {e}")
            return False
    
    async def delete(self, key: str) -> bool:
        """刪除緩存"""
        try:
            result = self.redis_client.delete(key)
            return result > 0
        except Exception as e:
            log.error(f"Cache delete error for key {key}: {e}")
            return False
    
    async def get_or_set(self, key: str, default_func, ttl: Optional[int] = None) -> Any:
        """獲取或設置緩存"""
        cached = await self.get(key)
        if cached is not None:
            return cached
        
        value = default_func()
        await self.set(key, value, ttl)
        return value
    
    async def clear_pattern(self, pattern: str) -> int:
        """清除匹配模式的緩存"""
        try:
            keys = self.redis_client.keys(pattern)
            if keys:
                return self.redis_client.delete(*keys)
            return 0
        except Exception as e:
            log.error(f"Cache clear pattern error: {e}")
            return 0

# 全局緩存實例
cache = CacheManager()

from fastapi import HTTPException, status
from app.core.cache import cache
from app.core.logging import log
import time
from typing import Optional

class RateLimiter:
    """速率限制器"""
    
    def __init__(self, requests_per_minute: int = 60):
        self.requests_per_minute = requests_per_minute
    
    async def check_rate_limit(self, identifier: str, cost: int = 1) -> bool:
        """檢查速率限制"""
        cache_key = f"rate_limit:{identifier}"
        
        # 獲取當前窗口數(shù)據(jù)
        window_data = await cache.get(cache_key) or {
            'count': 0,
            'window_start': time.time()
        }
        
        current_time = time.time()
        window_duration = 60  # 60秒窗口
        
        # 如果窗口已過期，重置
        if current_time - window_data['window_start'] > window_duration:
            window_data = {
                'count': 0,
                'window_start': current_time
            }
        
        # 檢查是否超過限制
        if window_data['count'] + cost > self.requests_per_minute:
            log.warning(f"Rate limit exceeded for {identifier}")
            return False
        
        # 更新計數(shù)
        window_data['count'] += cost
        await cache.set(cache_key, window_data, window_duration)
        
        return True
    
    async def get_remaining_requests(self, identifier: str) -> int:
        """獲取剩余請求數(shù)"""
        cache_key = f"rate_limit:{identifier}"
        window_data = await cache.get(cache_key) or {
            'count': 0,
            'window_start': time.time()
        }
        
        return max(0, self.requests_per_minute - window_data['count'])

def rate_limit_middleware(requests_per_minute: int = 60):
    """速率限制中間件工廠"""
    limiter = RateLimiter(requests_per_minute)
    
    async def middleware(identifier: str, cost: int = 1):
        if not await limiter.check_rate_limit(identifier, cost):
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=f"Rate limit exceeded. Try again in 60 seconds."
            )
        return True
    
    return middleware

# 默認速率限制器
default_limiter = RateLimiter()

import pytest
import asyncio
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.main import app
from app.models.database import Base, get_db

# 測試數(shù)據(jù)庫配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,
)

TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture(scope="function")
def db_session():
    """創(chuàng)建測試數(shù)據(jù)庫會話"""
    Base.metadata.create_all(bind=engine)
    session = TestingSessionLocal()
    try:
        yield session
    finally:
        session.close()
    Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def client(db_session):
    """創(chuàng)建測試客戶端"""
    def override_get_db():
        try:
            yield db_session
        finally:
            pass
    
    app.dependency_overrides[get_db] = override_get_db
    with TestClient(app) as test_client:
        yield test_client
    app.dependency_overrides.clear()

@pytest.fixture(scope="session")
def event_loop():
    """創(chuàng)建事件循環(huán)"""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

import pytest
from app.models.schemas import ChatRequest

def test_health_check(client):
    """測試健康檢查端點"""
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"

def test_chat_endpoint(client):
    """測試聊天端點"""
    # 注意：需要先設置認證或跳過認證
    request_data = {
        "message": "Hello, test message",
        "stream": False
    }
    
    response = client.post("/api/v1/chat", json=request_data)
    # 由于認證，這里可能會返回401
    assert response.status_code in [200, 401, 403]

def test_chat_streaming(client):
    """測試流式聊天"""
    request_data = {
        "message": "Hello, streaming test",
        "stream": True
    }
    
    response = client.post("/api/v1/chat", json=request_data)
    assert response.status_code in [200, 401, 403]

@pytest.mark.asyncio
async def test_tool_registry():
    """測試工具注冊表"""
    from app.tools.registry import tool_registry
    from app.tools.base import CalculatorTool
    
    # 測試工具注冊
    calculator = CalculatorTool()
    tool_registry.register_tool(calculator)
    
    # 測試工具獲取
    tool = tool_registry.get_tool("calculator")
    assert tool is not None
    assert tool.name == "calculator"

def test_metrics_endpoint(client):
    """測試指標端點"""
    response = client.get("/metrics")
    assert response.status_code == 200
    assert "http_requests_total" in response.text

version: '3.8'

services:
  ai-agent:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DEBUG=false
      - DATABASE_URL=postgresql://user:pass@db:5432/ai_agent
      - REDIS_URL=redis://redis:6379
      - SECRET_KEY=your-production-secret-key
    depends_on:
      - db
      - redis
    volumes:
      - ./logs:/app/logs
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  db:
    image: postgres:13
    environment:
      - POSTGRES_DB=ai_agent
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=pass
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./scripts/init-db.sql:/docker-entrypoint-initdb.d/init.sql
    restart: unless-stopped
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d ai_agent"]
      interval: 30s
      timeout: 10s
      retries: 3

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 30s
      timeout: 10s
      retries: 3

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - ai-agent
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:

events {
worker_connections 1024;
}

http {
upstream ai_agent {
server ai-agent:8000;
}

server {
    listen 80;
    server_name your-domain.com;
    return 301 https://$server_name$request_uri;
}

server {
    listen 443 ssl http2;
    server_name your-domain.com;

    ssl_certificate /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;

    # 安全頭
    add_header X-Frame-Options DENY;
    add_header X-Content-Type-Options nosniff;
    add_header X-XSS-Protection "1; mode=block";

    # 速率限制
    limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;

    location / {
        limit_req zone=api burst=20 nodelay;
        proxy_pass http://ai_agent;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    location /metrics {
        # 內(nèi)部訪問指標
        allow 127.0.0.1;
        deny all;
        proxy_pass http://ai_agent;
    }

    location /health {
        proxy_pass http://ai_agent;
        access_log off;
    }
}
}

# 應用配置
DEBUG=false
APP_NAME="AI Agent Service"
VERSION=1.0.0

# 安全配置
SECRET_KEY=your-super-secret-key-here-change-in-production
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

# 數(shù)據(jù)庫配置
DATABASE_URL=postgresql://user:pass@localhost:5432/ai_agent
REDIS_URL=redis://localhost:6379

# LLM配置
OPENAI_API_KEY=your-openai-api-key
OPENAI_BASE_URL=https://api.openai.com/v1
MODEL_NAME=gpt-3.5-turbo

# 速率限制
RATE_LIMIT_PER_MINUTE=60

#!/bin/bash

# AI Agent服務啟動腳本

set -e

echo "?? Starting AI Agent Service..."

# 檢查環(huán)境變量
if [ -z "$DATABASE_URL" ]; then
    echo "? DATABASE_URL is not set"
    exit 1
fi

if [ -z "$OPENAI_API_KEY" ]; then
    echo "??  OPENAI_API_KEY is not set, some features may not work"
fi

# 等待數(shù)據(jù)庫就緒
echo "? Waiting for database..."
while ! nc -z $(echo $DATABASE_URL | sed 's/.*@//' | cut -d: -f1) $(echo $DATABASE_URL | sed 's/.*://' | cut -d/ -f1); do
    sleep 1
done

echo "? Database is ready"

# 運行數(shù)據(jù)庫遷移
echo "?? Running database migrations..."
python -c "
from app.models.database import create_tables
create_tables()
print('Database tables created/verified')
"

# 啟動服務
echo "?? Starting Uvicorn server..."
exec uvicorn app.main:app \
    --host 0.0.0.0 \
    --port 8000 \
    --workers 4 \
    --log-level info

#!/bin/bash

# 部署腳本

set -e

ENV=${1:-staging}
IMAGE_TAG="ai-agent-service:latest"

echo "?? Deploying AI Agent Service to $ENV"

# 構建Docker鏡像
echo "?? Building Docker image..."
docker build -t $IMAGE_TAG .

# 根據(jù)環(huán)境選擇配置
if [ "$ENV" = "production" ]; then
    COMPOSE_FILE="docker-compose.prod.yml"
else
    COMPOSE_FILE="docker-compose.yml"
fi

# 部署服務
echo "?? Deploying services..."
docker-compose -f $COMPOSE_FILE down
docker-compose -f $COMPOSE_FILE up -d

echo "? Deployment completed!"
echo "?? Health check: curl http://localhost/health"

# 安裝依賴

pip install -r requirements.txt

# 啟動開發(fā)服務器

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 開發(fā)環(huán)境

docker-compose up -d

# 生產(chǎn)環(huán)境

docker-compose -f docker-compose.prod.yml up -d

# 克隆項目
git clone <your-repo>
cd ai-agent-service

# 設置環(huán)境
cp .env.example .env
# 編輯.env文件配置你的設置

# 方式1: 使用Docker（推薦）
docker-compose up -d

???????# 方式2: 本地開發(fā)
pip install -r requirements.txt
uvicorn app.main:app --reload