llm-app-development

大模型应用开发

概述

本技能涵盖构建由大模型驱动的应用的全方位内容，包括检索增强生成（RAG）流水线、向量数据库集成、提示词工程技巧、结构化输出生成、工具使用与 Agent 模式、评估框架、成本优化以及流式响应处理。

构建聊天机器人、AI 助手、知识检索系统、内容生成工具、自主 Agent，或任何将大模型能力集成到核心功能中的应用时，请使用本技能。

---

核心原则

1. 检索优于记忆——使用 RAG 将大模型的回答建立在真实数据之上，而不是依赖模型的参数记忆。这有助于减少幻觉，并确保答案保持最新。
2. 结构化 I/O 边界——为输入（系统提示词、上下文）和输出（通过函数调用或 Zod 模式定义的类型化响应）定义严格的模式。绝不要将原始大模型文本用于下游逻辑。
3. 上线前先评估——每项大模型功能都需要自动化评估。模型输出具有非确定性；没有评估，就无法衡量回归或改进。
4. 成本感知型架构——Token 使用量决定成本。积极使用缓存，选择满足质量要求的最小模型，并尽可能批量处理。
5. 优雅地处理失败——大模型可能拒绝回答、产生幻觉或超时。每条调用路径都需要回退行为、重试逻辑和用户可见的错误状态。

---

关键模式

模式 1：RAG 流水线架构

适用场景： 大模型需要访问超出上下文窗口限制的私有、大量或频繁更新的知识时。

实现：

import { OpenAIEmbeddings } from "@langchain/openai";
import { PGVectorStore } from "@langchain/community/vectorstores/pgvector";
import { RecursiveCharacterTextSplitter } from "langchain/text_splitter";

// 1. Chunking - Split documents into retrieval-friendly segments
const splitter = new RecursiveCharacterTextSplitter({
  chunkSize: 1000,
  chunkOverlap: 200,
  separators: ["\n\n", "\n", ". ", " "],
});

const chunks = await splitter.splitDocuments(documents);

// 2. Embedding - Convert chunks to vectors
const embeddings = new OpenAIEmbeddings({
  model: "text-embedding-3-small",
  dimensions: 1536,
});

// 3. Storage - Index in vector database
const vectorStore = await PGVectorStore.fromDocuments(chunks, embeddings, {
  postgresConnectionOptions: {
    connectionString: process.env.DATABASE_URL,
  },
  tableName: "documents",
  columns: {
    idColumnName: "id",
    vectorColumnName: "embedding",
    contentColumnName: "content",
    metadataColumnName: "metadata",
  },
});

// 4. Retrieval - Find relevant context for a query
async function retrieve(query: string, k: number = 5) {
  const results = await vectorStore.similaritySearchWithScore(query, k);

  // Filter by relevance threshold
  return results
    .filter(([_, score]) => score > 0.7)
    .map(([doc]) => doc);
}

// 5. Generation - Augment prompt with retrieved context
async function generateAnswer(query: string): Promise<string> {
  const context = await retrieve(query);

  const contextText = context
    .map((doc) => doc.pageContent)
    .join("\n---\n");

  const response = await openai.chat.completions.create({
    model: "gpt-4o",
    messages: [
      {
        role: "system",
        content: `Answer questions using ONLY the provided context. If the context doesn't contain the answer, say "I don't have information about that."

Context:
${contextText}`,
      },
      { role: "user", content: query },
    ],
    temperature: 0.1,
  });

  return response.choices[0].message.content ?? "";
}

原因： RAG 将知识存储与推理分离。大模型专注于整合答案，而向量数据库则负责大规模检索。这种架构支持在不重新训练的情况下更新知识，并且只纳入相关上下文，从而控制 Token 成本。

---

模式 2：使用 Zod 模式生成结构化输出

适用场景： 大模型输出将用于需要类型化、经过验证的数据的下游逻辑、API 或 UI 渲染时。

实现：

import { z } from "zod";
import OpenAI from "openai";
import { zodResponseFormat } from "openai/helpers/zod";

// Define the output schema
const ProductReviewAnalysis = z.object({
  sentiment: z.enum(["positive", "negative", "neutral", "mixed"]),
  confidence: z.number().min(0).max(1),
  themes: z.array(z.object({
    name: z.string(),
    sentiment: z.enum(["positive", "negative", "neutral"]),
    mentions: z.number(),
  })),
  summary: z.string().max(200),
  actionItems: z.array(z.string()),
});

type ProductReviewAnalysis = z.infer<typeof ProductReviewAnalysis>;

const openai = new OpenAI();

async function analyzeReviews(
  reviews: string[]
): Promise<ProductReviewAnalysis> {
  const response = await openai.beta.chat.completions.parse({
    model: "gpt-4o-2024-08-06",
    messages: [
      {
        role: "system",
        content: "Analyze product reviews and extract structured insights.",
      },
      {
        role: "user",
        content: `Analyze these reviews:\n${reviews.join("\n---\n")}`,
      },
    ],
    response_format: zodResponseFormat(ProductReviewAnalysis, "review_analysis"),
  });

  const parsed = response.choices[0].message.parsed;
  if (!parsed) {
    throw new Error("Failed to parse structured output");
  }
  return parsed;
}

原因： 结构化输出避免了对大模型文本进行脆弱的正则表达式解析。模型会被约束为生成符合你定义模式的有效 JSON，从而提供类型安全的数据，用于渲染 UI 组件、存入数据库或传递给其他服务。

---

模式 3：工具使用与 AI Agent

适用场景： 大模型需要采取行动（搜索、计算、调用 API、修改数据），而不只是生成文本时。

实现：

import OpenAI from "openai";

const tools: OpenAI.Chat.Completions.ChatCompletionTool[] = [
  {
    type: "function",
    function: {
      name: "search_knowledge_base",
      description: "Search the internal knowledge base for relevant articles",
      parameters: {
        type: "object",
        properties: {
          query: { type: "string", description: "Search query" },
          category: {
            type: "string",
            enum: ["billing", "technical", "account"],
            description: "Category to filter by",
          },
        },
        required: ["query"],
      },
    },
  },
  {
    type: "function",
    function: {
      name: "create_support_ticket",
      description: "Create a support ticket for unresolved issues",
      parameters: {
        type: "object",
        properties: {
          title: { type: "string" },
          description: { type: "string" },
          priority: { type: "string", enum: ["low", "medium", "high", "urgent"] },
        },
        required: ["title", "description", "priority"],
      },
    },
  },
];

// Tool implementations
const toolHandlers: Record<string, (args: unknown) => Promise<string>> = {
  search_knowledge_base: async (args) => {
    const { query, category } = args as { query: string; category?: string };
    const results = await knowledgeBase.search(query, { category });
    return JSON.stringify(results);
  },
  create_support_ticket: async (args) => {
    const ticket = args as { title: string; description: string; priority: string };
    const created = await ticketSystem.create(ticket);
    return JSON.stringify({ ticketId: created.id, status: "created" });
  },
};

// Agentic loop - let the model decide which tools to call
async function agentLoop(
  messages: OpenAI.Chat.Completions.ChatCompletionMessageParam[]
): Promise<string> {
  const MAX_ITERATIONS = 10;

  for (let i = 0; i < MAX_ITERATIONS; i++) {
    const response = await openai.chat.completions.create({
      model: "gpt-4o",
      messages,
      tools,
      tool_choice: "auto",
    });

    const message = response.choices[0].message;
    messages.push(message);

    // If no tool calls, we have the final answer
    if (!message.tool_calls || message.tool_calls.length === 0) {
      return message.content ?? "";
    }

    // Execute each tool call
    for (const toolCall of message.tool_calls) {
      const handler = toolHandlers[toolCall.function.name];
      if (!handler) {
        throw new Error(`Unknown tool: ${toolCall.function.name}`);
      }

      const args = JSON.parse(toolCall.function.arguments);
      const result = await handler(args);

      messages.push({
        role: "tool",
        tool_call_id: toolCall.id,
        content: result,
      });
    }
  }

  throw new Error("Agent exceeded maximum iterations");
}

原因： 工具使用将大模型从文本生成器转变为能够查询数据库、调用 API 和编排工作流的执行者。Agent 循环模式赋予模型自主决定使用哪些工具及何时使用的能力，同时迭代次数限制可防止失控执行。

---

模式 4：通过语义缓存优化成本

适用场景： 类似查询很常见，且对数据新鲜度的要求允许使用缓存时。

实现：

import { Redis } from "ioredis";
import { createHash } from "crypto";

interface CacheEntry {
  response: string;
  embedding: number[];
  createdAt: number;
}

class SemanticCache {
  private redis: Redis;
  private ttlSeconds: number;
  private similarityThreshold: number;

  constructor(options: {
    redisUrl: string;
    ttlSeconds?: number;
    similarityThreshold?: number;
  }) {
    this.redis = new Redis(options.redisUrl);
    this.ttlSeconds = options.ttlSeconds ?? 3600;
    this.similarityThreshold = options.similarityThreshold ?? 0.95;
  }

  // Exact match cache (fast, cheap)
  private exactKey(prompt: string): string {
    const hash = createHash("sha256").update(prompt).digest("hex");
    return `llm:exact:${hash}`;
  }

  // Check exact cache first, then semantic similarity
  async get(prompt: string): Promise<string | null> {
    // 1. Try exact match
    const exact = await this.redis.get(this.exactKey(prompt));
    if (exact) return exact;

    // 2. Try semantic match (more expensive, but catches paraphrases)
    const embedding = await getEmbedding(prompt);
    const candidates = await this.redis.smembers("llm:semantic:keys");

    for (const candidateKey of candidates) {
      const raw = await this.redis.get(`llm:semantic:${candidateKey}`);
      if (!raw) continue;

      const entry: CacheEntry = JSON.parse(raw);
      const similarity = cosineSimilarity(embedding, entry.embedding);

      if (similarity >= this.similarityThreshold) {
        return entry.response;
      }
    }

    return null;
  }

  async set(prompt: string, response: string): Promise<void> {
    const embedding = await getEmbedding(prompt);
    const key = createHash("sha256").update(prompt).digest("hex");

    // Store exact match
    await this.redis.setex(this.exactKey(prompt), this.ttlSeconds, response);

    // Store semantic entry
    const entry: CacheEntry = {
      response,
      embedding,
      createdAt: Date.now(),
    };
    await this.redis.setex(
      `llm:semantic:${key}`,
      this.ttlSeconds,
      JSON.stringify(entry)
    );
    await this.redis.sadd("llm:semantic:keys", key);
  }
}

function cosineSimilarity(a: number[], b: number[]): number {
  let dot = 0, magA = 0, magB = 0;
  for (let i = 0; i < a.length; i++) {
    dot += a[i] * b[i];
    magA += a[i] * a[i];
    magB += b[i] * b[i];
  }
  return dot / (Math.sqrt(magA) * Math.sqrt(magB));
}

原因： 大模型 API 调用成本高且速度慢。语义缓存不仅能命中完全相同的查询，还能命中改写后的查询，对于查询模式重复的应用（常见问题机器人、客户支持、搜索）可大幅降低成本。

---

模式 5：流式响应

适用场景： 任何对感知延迟有要求的面向用户的大模型交互。

实现：

// Server - Next.js Route Handler with streaming
import { OpenAI } from "openai";
import { NextRequest } from "next/server";

export async function POST(req: NextRequest) {
  const { messages } = await req.json();
  const openai = new OpenAI();

  const stream = await openai.chat.completions.create({
    model: "gpt-4o",
    messages,
    stream: true,
  });

  // Convert OpenAI stream to ReadableStream
  const encoder = new TextEncoder();
  const readable = new ReadableStream({
    async start(controller) {
      for await (const chunk of stream) {
        const text = chunk.choices[0]?.delta?.content ?? "";
        if (text) {
          controller.enqueue(encoder.encode(`data: ${JSON.stringify({ text })}\n\n`));
        }
      }
      controller.enqueue(encoder.encode("data: [DONE]\n\n"));
      controller.close();
    },
  });

  return new Response(readable, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    },
  });
}

// Client - React hook for consuming streams
function useStreamingChat() {
  const [messages, setMessages] = useState<Message[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);

  const sendMessage = async (content: string) => {
    const userMessage = { role: "user" as const, content };
    const updatedMessages = [...messages, userMessage];
    setMessages(updatedMessages);
    setIsStreaming(true);

    const response = await fetch("/api/chat", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ messages: updatedMessages }),
    });

    const reader = response.body?.getReader();
    const decoder = new TextDecoder();
    let assistantContent = "";

    setMessages((prev) => [...prev, { role: "assistant", content: "" }]);

    while (reader) {
      const { done, value } = await reader.read();
      if (done) break;

      const text = decoder.decode(value);
      const lines = text.split("\n").filter((line) => line.startsWith("data: "));

      for (const line of lines) {
        const data = line.slice(6);
        if (data === "[DONE]") break;

        const parsed = JSON.parse(data);
        assistantContent += parsed.text;

        setMessages((prev) => {
          const updated = [...prev];
          updated[updated.length - 1] = {
            role: "assistant",
            content: assistantContent,
          };
          return updated;
        });
      }
    }

    setIsStreaming(false);
  };

  return { messages, sendMessage, isStreaming };
}

原因： 如果没有流式响应，用户只能盯着空白屏幕等待 2–10 秒。流式响应会在 Token 到达时逐个显示，将首个 Token 的感知延迟降至 500 毫秒以内。这是任何面向用户的大模型功能的基本要求。

---

模式 6：大模型输出评估

适用场景： 在将任何大模型功能部署到生产环境之前，以及作为持续回归测试的一部分。

实现：

import { evaluate } from "braintrust";

// Define evaluation dataset
const dataset = [
  {
    input: "What is our refund policy?",
    expected: "30-day money-back guarantee",
    tags: ["policy", "refund"],
  },
  {
    input: "How do I reset my password?",
    expected: "Go to Settings > Security > Reset Password",
    tags: ["account", "password"],
  },
];

// Custom scoring functions
function factualAccuracy(output: string, expected: string): number {
  const expectedFacts = expected.toLowerCase().split(/[,.]/).map((s) => s.trim());
  const matchedFacts = expectedFacts.filter((fact) =>
    output.toLowerCase().includes(fact)
  );
  return matchedFacts.length / expectedFacts.length;
}

function answerRelevance(output: string, input: string): number {
  // Use an LLM as a judge
  // Returns 0-1 score for how relevant the answer is to the question
  return llmJudge(input, output, "relevance");
}

// Run evaluation
const results = await evaluate("support-bot-v2", {
  data: dataset,
  task: async (input) => {
    return await generateAnswer(input.input);
  },
  scores: [
    {
      name: "factual_accuracy",
      scorer: (args) => factualAccuracy(args.output, args.expected),
    },
    {
      name: "answer_relevance",
      scorer: (args) => answerRelevance(args.output, args.input),
    },
    {
      name: "no_hallucination",
      scorer: (args) => {
        const hallucinationCheck = detectHallucination(args.output, args.context);
        return hallucinationCheck ? 0 : 1;
      },
    },
  ],
});

console.log(`Average factual accuracy: ${results.scores.factual_accuracy}`);
console.log(`Average relevance: ${results.scores.answer_relevance}`);

原因： 大模型的输出具有概率性。如果不进行评估，就无法衡量提示词变更、模型升级或上下文修改是否提升或降低了质量。评估是大模型工程中的测试。

---

向量数据库选型指南

数据库	最适合	扩展方式	托管服务	本地开发
pgvector	现有 Postgres 用户，少于 1000 万个向量	垂直扩展	通过 Supabase/Neon	Docker
Pinecone	无服务器、大规模生产环境	水平扩展	是（仅提供托管服务）	模拟客户端
Chroma	原型开发、小型数据集	有限	Chroma Cloud	内存中
Weaviate	混合搜索（向量 + BM25）	水平扩展	提供托管服务和可自托管选项	Docker
Qdrant	对性能要求高、需要过滤	水平扩展	提供托管服务和可自托管选项	Docker

---

提示词工程快速参考

技术	适用场景	示例前缀
零样本	简单、定义明确的任务	“将这封电子邮件分类为垃圾邮件或非垃圾邮件：”
少样本	需要演示模式	“以下是一些示例：\n...”
模型思考	推理任务	启用提供方的思考/努力程度设置，而不是使用“逐步思考”前缀
系统提示词	持续生效的行为规则	role: "system" 消息
分隔符	将上下文与指令分开	"""context""" 或 <context> 标签

---

反模式

反模式	不良原因	更好的做法
将整份文档塞入上下文	浪费令牌、削弱相关性	使用带有分块和检索的 RAG
用正则表达式解析大模型文本输出	脆弱，格式变更时容易失效	使用结构化输出（函数调用、Zod）
生产部署前不做评估	无法衡量质量或发现回归问题	使用评分函数构建评估套件
用一个庞大的提示词处理所有任务	难以调试、成本高、速度慢	拆分为更小、更聚焦的提示词
开发期间忽略令牌成本	规模扩大后账单超出预期	跟踪每次查询的成本，设置预算并进行缓存
在请求处理路径中同步调用大模型	用户体验缓慢、有超时风险	流式传输响应；对于耗时任务，使用后台作业
到处硬编码模型名称	升级或进行 A/B 测试时很麻烦	通过配置驱动模型选择
API 调用没有重试逻辑	瞬时故障会导致用户可见的错误	使用带抖动的指数退避

---

检查清单

• RAG 流程：已选择并测试分块策略（大小、重叠、分隔符）
• 已选择向量数据库并配置连接池
• 已选择嵌入模型（维度、成本、质量之间的权衡）
• 已为所有大模型到代码的边界定义结构化输出模式
• 已为面向用户的响应实现流式传输
• 已建立评估套件，每项功能至少包含 20 个测试用例
• 已为重复查询启用语义缓存或精确缓存
• 错误处理：重试、回退方案、用户可见的错误状态
• 成本跟踪：记录每次查询的令牌用量
• 已对大模型 API 调用实施速率限制
• 系统提示词已进行版本管理，并存储在代码之外
• 提示词注入防护（输入验证、输出过滤）

---

相关资源

• 技能： application-security（提示词注入）、monitoring-observability（大模型调用跟踪）
• 规则： docs/reference/stacks/react-typescript.md（聊天 UI 的前端模式）
• 规则： docs/reference/checklists/verification-template.md（评估检查清单）