内部 AI 工具产品化：从脚本到可运营能力

// ❌ 坏做法：硬编码 Key，所有人共用
const OPENAI_API_KEY = 'sk-xxxxxx'
 
async function summarize(text: string) {
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${OPENAI_API_KEY}`,
      'Content-Type': 'application/json',
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages: [{ role: 'user', content: text }],
    }),
  })
  return response.json()
}

// ✅ 好做法：按用户身份分配 Key，带审计标签
interface SummarizeParams {
  text: string
  userId: string
  teamId: string
}
 
async function summarize({ text, userId, teamId }: SummarizeParams) {
  // 从 Vault 按团队获取独立 Key，而不是全局共享
  const apiKey = await vault.getSecret(`openai/${teamId}`)
 
  // 检查该团队本月剩余额度
  const budget = await budgetService.checkRemaining(teamId)
  if (budget < estimateCost(text)) {
    throw new BudgetExceededError(`团队 ${teamId} 本月 AI 额度已用完`)
  }
 
  // 输入内容脱敏：移除手机号、身份证号等 PII
  const sanitizedText = await piiFilter.sanitize(text)
 
  const response = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Authorization': `Bearer ${apiKey}`,
      'Content-Type': 'application/json',
      // 在 Header 里带上调用方身份，方便审计溯源
      'X-Caller-Id': userId,
      'X-Team-Id': teamId,
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages: [{ role: 'user', content: sanitizedText }],
    }),
  })
 
  // 记录调用日志到集中式审计系统
  await auditLog.record({
    userId,
    teamId,
    model: 'gpt-4',
    inputTokens: estimateTokens(sanitizedText),
    timestamp: new Date().toISOString(),
  })
 
  return response.json()
}

# ❌ 坏做法：调用完就完了，成本是个黑盒
import openai
 
def analyze_competitor(url: str):
    content = fetch_article(url)
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是一个竞品分析师"},
            {"role": "user", "content": f"请分析以下文章：\n{content}"},
        ],
    )
    return response.choices[0].message.content

# ✅ 好做法：每次调用都带业务标签，成本可归因
import openai
from dataclasses import dataclass
 
@dataclass
class CostAttribution:
    team: str          # 业务线
    feature: str       # 功能名称
    triggered_by: str  # 触发人
    project_id: str    # 关联项目
 
def analyze_competitor(url: str, attribution: CostAttribution):
    content = fetch_article(url)
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "你是一个竞品分析师"},
            {"role": "user", "content": f"请分析以下文章：\n{content}"},
        ],
    )
 
    # 从 API 响应中提取真实的 token 用量
    usage = response.usage
    cost = calculate_cost(
        model="gpt-4",
        prompt_tokens=usage.prompt_tokens,
        completion_tokens=usage.completion_tokens,
    )
 
    # 写入成本归因表，而不是只打到 stdout
    cost_tracker.record(
        team=attribution.team,
        feature=attribution.feature,
        triggered_by=attribution.triggered_by,
        model="gpt-4",
        prompt_tokens=usage.prompt_tokens,
        completion_tokens=usage.completion_tokens,
        cost_usd=cost,
        project_id=attribution.project_id,
        timestamp=datetime.utcnow(),
    )
 
    return response.choices[0].message.content

// ❌ 坏做法：catch 了异常但只 console.log，用户看到空白页
async function generateReply(ticket: Ticket) {
  try {
    const reply = await callAI(ticket.content)
    return reply
  } catch (error) {
    console.log('AI 调用失败', error)
    return null  // 调用方拿到 null，不知道该展示什么
  }
}

// ✅ 好做法：分级处理，用户看到有意义的反馈
interface AIResponse {
  content: string | null
  fallback: string      // 降级方案
  errorType?: ErrorType // 结构化的错误类型
}
 
async function generateReply(ticket: Ticket): Promise<AIResponse> {
  try {
    const reply = await callAI(ticket.content)
    return { content: reply, fallback: '' }
  } catch (error) {
    // 区分可重试和不可重试的错误
    if (isRateLimitError(error)) {
      // 限流：排队重试，用户无感
      await retryQueue.enqueue(ticket.id)
      return {
        content: null,
        fallback: 'AI 正在排队生成回复，预计 30 秒后刷新',
        errorType: 'RATE_LIMITED',
      }
    }
 
    if (isContentFilterError(error)) {
      // 内容被过滤：提示用户修改输入
      return {
        content: null,
        fallback: '输入内容可能包含敏感信息，请修改后重试',
        errorType: 'CONTENT_FILTERED',
      }
    }
 
    // 未知错误：记录到监控，给用户兜底提示
    await sentry.captureException(error, { tags: { ticketId: ticket.id } })
    return {
      content: null,
      fallback: 'AI 暂时无法生成回复，你可以先手动处理',
      errorType: 'UNKNOWN',
    }
  }
}

# ❌ 坏做法：日志打到 stdout，出了问题 grep 到天荒地老
def handle_request(user_input: str):
    print(f"收到请求: {user_input[:50]}")
    result = call_model(user_input)
    print(f"返回结果: {result[:50]}")
    return result

# ✅ 好做法：结构化日志，带 trace_id，可检索可聚合
import structlog
import uuid
 
logger = structlog.get_logger()
 
def handle_request(user_input: str, user_id: str, feature: str):
    trace_id = str(uuid.uuid4())
 
    logger.info(
        "ai_request_start",
        trace_id=trace_id,
        user_id=user_id,
        feature=feature,
        input_length=len(user_input),
    )
 
    try:
        result = call_model(user_input)
        logger.info(
            "ai_request_success",
            trace_id=trace_id,
            output_length=len(result),
            latency_ms=elapsed_ms,
        )
        return result
    except Exception as e:
        logger.error(
            "ai_request_failed",
            trace_id=trace_id,
            error_type=type(e).__name__,
            error_message=str(e),
        )
        raise