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):
    """應(yīng)用配置類"""
    
    # 應(yīng)用配置
    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ù)庫(kù)配置
    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():
    """配置結(jié)構(gòu)化日志"""
    
    # 移除默認(rèn)處理器
    logger.remove()
    
    # 添加控制臺(tái)處理器
    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實(shí)例
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):
    """聊天請(qǐng)求模型"""
    message: str = Field(..., description="用戶消息")
    conversation_id: Optional[str] = Field(None, description="會(huì)話ID")
    tools: Optional[List[str]] = Field(None, description="要使用的工具列表")
    stream: bool = Field(False, description="是否流式響應(yīng)")
    
    class Config:
        json_schema_extra = {
            "example": {
                "message": "你好，請(qǐng)幫我查詢天氣",
                "conversation_id": "conv_123",
                "tools": ["weather", "calculator"],
                "stream": False
            }
        }

class ChatResponse(BaseModel):
    """聊天響應(yīng)模型"""
    message: str = Field(..., description="AI回復(fù)")
    conversation_id: str = Field(..., description="會(huì)話ID")
    tool_calls: Optional[List[Dict[str, Any]]] = Field(None, description="工具調(diào)用")
    usage: Optional[Dict[str, int]] = Field(None, description="使用統(tǒng)計(jì)")
    
    class Config:
        json_schema_extra = {
            "example": {
                "message": "你好！我是AI助手，很高興為您服務(wù)。",
                "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):
    """工具響應(yīng)模型"""
    tool_name: str = Field(..., description="工具名稱")
    result: Any = Field(..., description="工具執(zhí)行結(jié)果")
    success: bool = Field(..., description="執(zhí)行是否成功")
    error_message: Optional[str] = Field(None, description="錯(cuò)誤信息")

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

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

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

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):
    """應(yīng)用生命周期管理"""
    # 啟動(dòng)時(shí)執(zhí)行
    log.info("?? Starting AI Agent Service")
    yield
    # 關(guān)閉時(shí)執(zhí)行
    log.info("?? Shutting down AI Agent Service")

app = FastAPI(
    title=settings.PROJECT_NAME,
    version=settings.VERSION,
    description="企業(yè)級(jí)AI Agent微服務(wù)",
    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):
    """添加請(qǐng)求處理時(shí)間頭"""
    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

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

# 注冊(cè)路由
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)
):
    """處理聊天請(qǐng)求"""
    try:
        log.info(f"Processing chat request for conversation: {request.conversation_id}")
        
        if request.stream:
            # 流式響應(yīng)處理
            return StreamingResponse(
                chat_service.stream_chat(request),
                media_type="text/event-stream"
            )
        else:
            # 普通響應(yīng)
            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):
    """獲取會(huì)話歷史"""
    # 實(shí)現(xiàn)獲取會(huì)話邏輯
    return {"conversation_id": conversation_id, "messages": []}

@router.delete("/conversations/{conversation_id}")
async def delete_conversation(conversation_id: str):
    """刪除會(huì)話"""
    # 實(shí)現(xiàn)刪除會(huì)話邏輯
    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:
    """聊天 服務(wù)"""
    
    def __init__(self, llm_service: LLMService):
        self.llm_service = llm_service
    
    async def process_chat(self, request: ChatRequest) -> ChatResponse:
        """處理聊天請(qǐng)求"""
        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]:
        """流式聊天響應(yīng)"""
        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服務(wù)基類"""
    
    async def generate_response(
        self, 
        message: str, 
        conversation_id: Optional[str] = None,
        tools: Optional[List[str]] = None
    ) -> Dict[str, Any]:
        """生成響應(yīng)"""
        raise NotImplementedError
    
    async def stream_response(
        self, 
        message: str, 
        conversation_id: Optional[str] = None,
        tools: Optional[List[str]] = None
    ) -> AsyncGenerator[str, None]:
        """流式生成響應(yīng)"""
        raise NotImplementedError

class OpenAIService(LLMService):
    """OpenAI服務(wù)實(shí)現(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):
    """計(jì)算器工具"""
    
    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í)行計(jì)算"""
        try:
            # 安全評(píng)估數(shù)學(xué)表達(dá)式
            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:
    """工具注冊(cè)表"""
    
    def __init__(self):
        self._tools: Dict[str, BaseTool] = {}
    
    def register_tool(self, tool: BaseTool):
        """注冊(cè)工具"""
        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)

# 全局工具注冊(cè)表實(shí)例
tool_registry = ToolRegistry()

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

# 初始化時(shí)注冊(cè)工具
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:
    """驗(yàn)證密碼"""
    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)建訪問(wèn)令牌"""
    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]:
    """驗(yàn)證令牌"""
    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)):
    """獲取當(dāng)前用戶"""
    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"},
        )
    
    # 這里可以添加用戶驗(yàn)證邏輯
    return payload

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

def get_chat_service(llm_service: LLMService = Depends(get_llm_service)) -> ChatService:
    """獲取聊天 服務(wù)實(shí)例"""
    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/*

# 復(fù)制依賴文件
COPY requirements.txt .

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

# 復(fù)制應(yīng)用代碼
COPY . .

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

# 暴露端口
EXPOSE 8000

# 啟動(dòng)命令
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)
    
    # 關(guān)系
    conversations = relationship("Conversation", back_populates="user")
    api_keys = relationship("APIKey", back_populates="user")

class Conversation(Base):
    """會(huì)話模型"""
    __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)
    
    # 關(guān)系
    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)
    
    # 關(guān)系
    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)
    
    # 關(guān)系
    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ù)庫(kù)引擎和會(huì)話
engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

def create_tables():
    """創(chuàng)建數(shù)據(jù)庫(kù)表"""
    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)
):
    """獲取用戶會(huì)話列表"""
    # 檢查權(quán)限
    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)計(jì)"""
    if not current_user.get("is_superuser", False):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    
    # 實(shí)現(xiàn)工具使用統(tǒng)計(jì)查詢
    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)
):
    """管理員刪除會(huì)話"""
    if not current_user.get("is_superuser", False):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required"
        )
    
    # 實(shí)現(xiàn)管理員刪除會(huì)話邏輯
    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

# 定義指標(biāo)
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ī)范化路徑用于指標(biāo)"""
        if path.startswith("/api/v1"):
            # 提取API版本后的第一部分作為端點(diǎn)
            parts = path.split("/")
            if len(parts) > 3:
                return f"/api/v1/{parts[3]}"
        return path

def metrics_endpoint(request: Request) -> Response:
    """提供Prometheus指標(biāo)"""
    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)

# 添加指標(biāo)端點(diǎn)（在路由部分）
@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小時(shí)默認(rèn)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:
        """設(shè)置緩存值"""
        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:
        """獲取或設(shè)置緩存"""
        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

# 全局緩存實(shí)例
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}"
        
        # 獲取當(dāng)前窗口數(shù)據(jù)
        window_data = await cache.get(cache_key) or {
            'count': 0,
            'window_start': time.time()
        }
        
        current_time = time.time()
        window_duration = 60  # 60秒窗口
        
        # 如果窗口已過(guò)期，重置
        if current_time - window_data['window_start'] > window_duration:
            window_data = {
                'count': 0,
                'window_start': current_time
            }
        
        # 檢查是否超過(guò)限制
        if window_data['count'] + cost > self.requests_per_minute:
            log.warning(f"Rate limit exceeded for {identifier}")
            return False
        
        # 更新計(jì)數(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:
        """獲取剩余請(qǐng)求數(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

# 默認(rèn)速率限制器
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

# 測(cè)試數(shù)據(jù)庫(kù)配置
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)建測(cè)試數(shù)據(jù)庫(kù)會(huì)話"""
    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)建測(cè)試客戶端"""
    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):
    """測(cè)試健康檢查端點(diǎn)"""
    response = client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"

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

def test_chat_streaming(client):
    """測(cè)試流式聊天"""
    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():
    """測(cè)試工具注冊(cè)表"""
    from app.tools.registry import tool_registry
    from app.tools.base import CalculatorTool
    
    # 測(cè)試工具注冊(cè)
    calculator = CalculatorTool()
    tool_registry.register_tool(calculator)
    
    # 測(cè)試工具獲取
    tool = tool_registry.get_tool("calculator")
    assert tool is not None
    assert tool.name == "calculator"

def test_metrics_endpoint(client):
    """測(cè)試指標(biāo)端點(diǎn)"""
    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)部訪問(wèn)指標(biāo)
        allow 127.0.0.1;
        deny all;
        proxy_pass http://ai_agent;
    }

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

# 應(yīng)用配置
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ù)庫(kù)配置
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服務(wù)啟動(dòng)腳本

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ù)庫(kù)就緒
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"

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

# 啟動(dòng)服務(wù)
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"

# 構(gòu)建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

# 部署服務(wù)
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

# 啟動(dòng)開發(fā)服務(wù)器

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

# 克隆項(xiàng)目
git clone <your-repo>
cd ai-agent-service

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

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

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