Building AI Features（產品中加入 AI）

// Streaming — Server-Sent Events (SSE)
app.MapPost("/api/chat", async (ChatRequest req, HttpContext context) =>
{
    // Input Guard
    if (await ContainsPromptInjection(req.Message))
        return Results.BadRequest("Invalid input");
 
    context.Response.ContentType = "text/event-stream";
    context.Response.Headers.CacheControl = "no-cache";
 
    // 選擇模型（成本控制）
    var model = req.Message.Length > 500 ? "claude-sonnet-4-20250514" : "claude-haiku-4-20250514";
 
    // Streaming
    await foreach (var token in StreamChatResponse(model, req.Message))
    {
        // Output Guard（逐步檢查）
        if (ContainsPII(token)) continue;
 
        await context.Response.WriteAsync($"data: {token}\n\n");
        await context.Response.Body.FlushAsync();
    }
    await context.Response.WriteAsync("data: [DONE]\n\n");
});
 
// Token 用量追蹤
public class UsageTracker
{
    public async Task TrackUsage(string userId, int inputTokens, int outputTokens, string model)
    {
        var cost = CalculateCost(inputTokens, outputTokens, model);
        await _db.UsageLogs.AddAsync(new UsageLog
        {
            UserId = userId,
            InputTokens = inputTokens,
            OutputTokens = outputTokens,
            Model = model,
            Cost = cost,
            Timestamp = DateTime.UtcNow
        });
    }
}

// Streaming with Anthropic SDK
import Anthropic from "@anthropic-ai/sdk";
 
const client = new Anthropic();
 
// SSE Endpoint
app.post("/api/chat", async (req, res) => {
  // Input Guard
  const input = req.body.message;
  if (await detectPromptInjection(input)) {
    return res.status(400).json({ error: "Invalid input" });
  }
 
  // 模型路由（成本控制）
  const model = selectModel(input);
 
  // SSE headers
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
 
  // Streaming
  const stream = await client.messages.stream({
    model,
    max_tokens: 1024,
    system: "你是一個產品助手。只回答與產品相關的問題。",
    messages: [{ role: "user", content: input }],
  });
 
  let totalTokens = 0;
 
  for await (const event of stream) {
    if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
      const text = event.delta.text;
      // Output Guard
      if (!containsSensitiveInfo(text)) {
        res.write(`data: ${JSON.stringify({ text })}\n\n`);
      }
    }
  }
 
  // 記錄用量
  const usage = await stream.finalMessage();
  await trackUsage(req.user.id, usage.usage.input_tokens, usage.usage.output_tokens, model);
 
  res.write("data: [DONE]\n\n");
  res.end();
});
 
// 模型路由
function selectModel(input: string): string {
  if (input.length < 100) return "claude-haiku-4-20250514";    // 簡單問題
  if (input.includes("分析") || input.includes("比較"))
    return "claude-sonnet-4-20250514"; // 複雜分析
  return "claude-haiku-4-20250514"; // 預設用便宜的
}
 
// 回應快取
const cache = new Map<string, string>();
 
async function getCachedOrGenerate(input: string): Promise<string> {
  const cacheKey = hashInput(input);
  if (cache.has(cacheKey)) return cache.get(cacheKey)!;
  const response = await generate(input);
  cache.set(cacheKey, response);
  return response;
}

import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import hashlib
from functools import lru_cache
 
app = FastAPI()
client = anthropic.Anthropic()
 
# Input Guard — Prompt Injection 偵測
INJECTION_PATTERNS = [
    "ignore previous instructions",
    "you are now",
    "system prompt",
    "reveal your instructions",
]
 
def detect_injection(text: str) -> bool:
    lower = text.lower()
    return any(pattern in lower for pattern in INJECTION_PATTERNS)
 
# Streaming Endpoint
@app.post("/api/chat")
async def chat(request: ChatRequest):
    if detect_injection(request.message):
        raise HTTPException(400, "Invalid input")
 
    # 模型路由
    model = select_model(request.message)
 
    async def generate():
        with client.messages.stream(
            model=model,
            max_tokens=1024,
            system="你是一個產品助手。只回答與產品相關的問題。如果問題與產品無關，禮貌拒絕。",
            messages=[{"role": "user", "content": request.message}],
        ) as stream:
            for text in stream.text_stream:
                # Output Guard
                if not contains_pii(text):
                    yield f"data: {text}\n\n"
            yield "data: [DONE]\n\n"
 
    return StreamingResponse(generate(), media_type="text/event-stream")
 
# 成本監控
class CostTracker:
    PRICES = {
        "claude-haiku-4-20250514": {"input": 0.25, "output": 1.25},     # per 1M tokens
        "claude-sonnet-4-20250514": {"input": 3.0, "output": 15.0},
    }
 
    def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
        prices = self.PRICES[model]
        return (input_tokens * prices["input"] + output_tokens * prices["output"]) / 1_000_000
 
    async def check_budget(self, user_id: str) -> bool:
        monthly_cost = await self.get_monthly_cost(user_id)
        return monthly_cost < 10.0  # 每個使用者每月上限 $10