自己接模型开发 AI 应用——底层逻辑全解

写给想搞清楚 LLM 应用开发本质的工程师。不讲玄学，只讲代码和流程。

本文用一个贯穿始终的例子：给博客平台做一个 AI 写作助手，从零到完整功能一步步实现。

---

核心认知（先记住这一句）

模型是一个无状态函数：f(messages[]) → text

它不认识你，没有记忆，不能主动做任何事。 状态、历史、数据、工具执行——全部由你的代码维护，每次调用都是全量传入。

后面所有内容都是这句话的展开。

---

Step 1：跑通第一个请求

目标：用户在博客编辑器里输入关键词，AI 返回一个标题建议。

安装依赖

npm install @anthropic-ai/sdk

最简实现

// lib/ai.ts
import Anthropic from '@anthropic-ai/sdk'

const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY  // 存到 .env.local，绝不硬编码
})

export async function generateTitle(keyword: string) {
  const response = await client.messages.create({
    model: 'claude-sonnet-4-6',
    max_tokens: 256,
    messages: [
      { role: 'user', content: `根据关键词"${keyword}"，给我 3 个吸引人的博客标题` }
    ]
  })

  return response.content[0].text
}

接口层

// app/api/ai/title/route.ts
import { generateTitle } from '@/lib/ai'

export async function POST(req: Request) {
  const { keyword } = await req.json()
  const titles = await generateTitle(keyword)
  return Response.json({ titles })
}

调用测试

curl -X POST http://localhost:3000/api/ai/title \
  -H "Content-Type: application/json" \
  -d '{"keyword": "Next.js 性能优化"}'

# 返回：
# { "titles": "1. 《让你的 Next.js 应用快 3 倍的 10 个技巧》\n2. ..." }

你刚才做了什么：一次 HTTP POST，发文字，收文字。AI 应用的本质就是这个。

---

Step 2：理解参数，控制模型行为

上面的代码能跑，但不够可控。先把参数搞清楚。

完整参数结构

const response = await client.messages.create({
  // ── 必填 ──────────────────────────────
  model:      'claude-sonnet-4-6',   // 用哪个模型
  max_tokens: 1024,                  // 输出最多多少 token
  messages:   [...],                 // 对话历史

  // ── 控制模型行为 ─────────────────────
  system:      '...',                // 幕后指令，用户看不到
  temperature: 0.7,                  // 随机性/创意度

  // ── 高级功能（用到再开）──────────────
  tools:   [...],                    // 工具定义（Tool Use）
  stream:  true,                     // 流式输出
})

model：怎么选？

模型	定位	适合场景
claude-opus-4-6	最强、最贵	复杂推理、高精度
claude-sonnet-4-6	性价比最高	日常首选
claude-haiku-4-5	最快、最便宜	简单任务、高并发

system：幕后规则

// 没有 system 的问题：模型什么都答，风格不可控
// 加了 system：模型被约束在你规定的范围内工作

system: `你是一个专业的中文博客写作助手。
规则：
- 只输出标题，不解释
- 每个标题不超过 20 个字
- 风格：实用、有数字、有价值感`

temperature：创意 vs 精确

0.0  → 每次输出几乎相同  →  代码生成、数据提取、格式转换
0.7  → 平衡              →  日常对话、内容生成（推荐默认值）
1.0  → 更有创意          →  头脑风暴、创意写作

max_tokens：输出上限

1 token ≈ 0.75 个英文单词 ≈ 0.5 个汉字

256   →  短回复、标题建议
1024  →  普通段落
4096  →  完整文章

超出就截断，不是保证输出这么多。

返回值：你需要关心的字段

const response = await client.messages.create({...})

response.content[0].text   // 模型的文字回答，最常用
response.stop_reason       // 'end_turn'(正常) | 'tool_use'(要调工具) | 'max_tokens'(被截断)
response.usage             // { input_tokens: 150, output_tokens: 300 }，计费依据

改进后的标题生成

export async function generateTitle(keyword: string) {
  const response = await client.messages.create({
    model: 'claude-sonnet-4-6',
    max_tokens: 256,
    temperature: 0.8,          // 标题需要创意，调高一点
    system: `你是专业博客标题写手。
输出格式：直接输出 3 个标题，每行一个，不加序号和解释。
风格：有数字、有价值感、适合 SEO。`,
    messages: [
      { role: 'user', content: `关键词：${keyword}` }
    ]
  })

  return response.content[0].text.split('\n').filter(Boolean)
  // → ['让你的 Next.js 快 3 倍的 10 个技巧', '...', '...']
}

---

Step 3：多轮对话——让 AI 记住上下文

目标：用户说"标题太长了"，AI 知道是在改哪个标题，而不是重新开始。

模型没有记忆，你来维护历史

第1轮发送：[你好]
第2轮发送：[你好, 好的有什么可以帮你, 帮我写标题]
第3轮发送：[你好, 好的有什么可以帮你, 帮我写标题, 这是3个标题, 改短一点]

每次请求都把完整历史带上，模型才能"记住"前面说了什么。

实现

// 用数组维护历史
type Message = { role: 'user' | 'assistant'; content: string }

export class BlogAIChat {
  private history: Message[] = []

  async send(userInput: string): Promise<string> {
    // 把用户输入加入历史
    this.history.push({ role: 'user', content: userInput })

    const response = await client.messages.create({
      model: 'claude-sonnet-4-6',
      max_tokens: 1024,
      system: '你是博客写作助手，帮助用户打磨文章标题和内容。',
      messages: this.history  // 每次发送完整历史
    })

    const reply = response.content[0].text

    // 把 AI 回复也加入历史，下轮才能看到
    this.history.push({ role: 'assistant', content: reply })

    return reply
  }

  clear() {
    this.history = []  // 开始新对话时清空
  }
}

对话效果

const chat = new BlogAIChat()

await chat.send('帮我写3个关于 Next.js 的标题')
// → "1. 《Next.js 15 新特性...》\n2. ..."

await chat.send('第一个太长了，控制在 15 字以内')
// → "《Next.js 15 必学新特性》"  ← 知道是在改第一个

await chat.send('换个角度，从性能优化切入')
// → "《Next.js 性能翻倍实战》"   ← 知道还是在改标题

注意：历史越长越贵

对话历史 10 轮 → input_tokens 可能高达 3000+
对话历史 50 轮 → input_tokens 可能高达 15000+

处理方式：
1. 超过 N 轮后，截掉最早的几轮
2. 让模型对历史做摘要，替换掉详细内容
3. 业务上限制每次对话长度

---

Step 4：流式输出——打字机效果

目标：AI 生成文章时，不是等全部写完才显示，而是实时一字一字出现。

为什么需要流式

不加流式：模型写 500 字的文章 → 用户等 8 秒 → 一次性全部显示
加流式：  模型写 500 字的文章 → 用户立刻看到第一个字 → 字符逐渐出现

用户体验差距极大，生产环境基本都要加流式。

后端：用 SSE 推给前端

// app/api/ai/write/route.ts
export async function POST(req: Request) {
  const { prompt } = await req.json()
  const encoder = new TextEncoder()

  const stream = new ReadableStream({
    async start(controller) {
      const aiStream = client.messages.stream({
        model: 'claude-sonnet-4-6',
        max_tokens: 2048,
        messages: [{ role: 'user', content: prompt }]
      })

      for await (const chunk of aiStream) {
        if (chunk.type === 'content_block_delta' && chunk.delta.type === 'text_delta') {
          // SSE 格式：data: 内容\n\n
          controller.enqueue(encoder.encode(`data: ${JSON.stringify({ text: chunk.delta.text })}\n\n`))
        }
      }

      controller.enqueue(encoder.encode('data: [DONE]\n\n'))
      controller.close()
    }
  })

  return new Response(stream, {
    headers: {
      'Content-Type':  'text/event-stream',
      'Cache-Control': 'no-cache',
    }
  })
}

前端：接收并实时展示

// components/AIWriter.tsx
async function startWriting(prompt: string) {
  let content = ''

  const response = await fetch('/api/ai/write', {
    method: 'POST',
    body: JSON.stringify({ prompt })
  })

  const reader = response.body!.getReader()
  const decoder = new TextDecoder()

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

    const lines = decoder.decode(value).split('\n')
    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = line.slice(6)
        if (data === '[DONE]') return

        const { text } = JSON.parse(data)
        content += text
        setEditorContent(content)  // 实时更新 UI
      }
    }
  }
}

---

Step 5：RAG——让模型知道你的私有数据

目标：用户问"帮我分析一下访问量最高的文章有什么共同特点"，AI 能基于真实数据回答。

问题根源

模型的知识 = 训练截止日期前的公开数据
           ≠ 你的数据库、用户数据、实时信息

解法：你查数据，把结果告诉模型。

方式一：直接注入 Prompt（适合小数据）

export async function analyzeBlogs(userId: string) {
  // 第一步：你来查数据库
  const blogs = await db.query(`
    SELECT title, views, avg_read_time, bounce_rate
    FROM blogs
    WHERE user_id = ? AND created_at > NOW() - INTERVAL 30 DAY
    ORDER BY views DESC
    LIMIT 10
  `, [userId])

  // 第二步：把数据拼进 prompt，告诉模型
  const response = await client.messages.create({
    model: 'claude-sonnet-4-6',
    max_tokens: 1024,
    messages: [{
      role: 'user',
      content: `
以下是我最近 30 天访问量最高的 10 篇文章数据：

${JSON.stringify(blogs, null, 2)}

请分析这些高访问量文章的共同特点，给出 3 条写作建议。
      `
    }]
  })

  return response.content[0].text
}

// 模型拿到真实数据后的回答：
"根据你的数据分析，高访问量文章有以下共同特点：
1. 标题包含数字（'10个'、'3种'），点击率更高
2. 平均阅读时间在 4-6 分钟，说明内容深度合适
3. 跳出率低于 40% 的文章均有清晰的目录结构..."

方式二：向量检索（适合大量文档）

当数据量大（几百篇文章、长文档），不可能全塞进 prompt，用向量检索精准召回相关内容。

原理：

文本 → 向量（一串数字） → 相似文本的向量距离近

"苹果手机"  → [0.8, 0.2, 0.1, ...]
"iPhone"    → [0.79, 0.21, 0.09, ...]  ← 语义相似，向量接近
"香蕉"      → [0.1, 0.9, 0.3, ...]    ← 语义不同，向量远

建库阶段（一次性）：

import { OpenAI } from 'openai'  // 用 OpenAI 的 embedding API 举例

async function buildIndex(blogs: Blog[]) {
  for (const blog of blogs) {
    // 把文章内容转成向量
    const embedding = await openai.embeddings.create({
      model: 'text-embedding-3-small',
      input: blog.content
    })

    // 存入向量数据库（如 pgvector、Pinecone）
    await vectorDB.insert({
      id:        blog.id,
      vector:    embedding.data[0].embedding,
      metadata:  { title: blog.title, content: blog.content }
    })
  }
}

查询阶段（每次对话）：

async function ragQuery(userQuestion: string) {
  // 1. 把用户问题也转成向量
  const questionEmbedding = await openai.embeddings.create({
    model: 'text-embedding-3-small',
    input: userQuestion
  })

  // 2. 找最相似的 3 篇文章
  const relatedBlogs = await vectorDB.search(
    questionEmbedding.data[0].embedding,
    { topK: 3 }
  )

  // 3. 把召回的文章内容注入 prompt
  const context = relatedBlogs.map(b => b.metadata.content).join('\n---\n')

  const response = await client.messages.create({
    messages: [{
      role: 'user',
      content: `
参考以下文章内容回答问题：

${context}

问题：${userQuestion}
      `
    }]
  })

  return response.content[0].text
}

两种方式怎么选

	直接注入	向量检索
数据量	< 50 条 / 文档短	> 50 条 / 文档长
实现难度	简单，直接拼字符串	复杂，需要向量数据库
成本	token 消耗多	token 消耗少
精准度	全量数据，不会漏	依赖检索质量

实践建议：先用直接注入跑通功能，有性能/成本问题再上向量检索。

---

Step 6：Tool Use——让模型主动调用你的函数

目标：用户说"帮我把访问量低于 100 的草稿文章，标题加上'[待优化]'前缀"，AI 自动查数据库、自动更新。

RAG vs Tool Use 的本质区别

RAG：      你主动查数据 → 告诉模型 → 模型分析
Tool Use：  模型决定查什么 → 告诉你去查 → 你执行 → 告诉模型结果 → 模型回答

RAG 是你喂给模型，Tool Use 是模型指挥你执行。

工具调用是模型的能力吗？

是，也不是。

• 模型能：识别什么时候需要工具，输出结构化的调用指令（JSON）
• 模型不能：真正连接数据库、执行代码、调用 API——这些都是你的代码做的

模型只是"点菜"，你来"上菜"。

完整流程（来回两次）

你                                    模型
─────────────────────────────────────────────
① 发请求（带工具定义）           →
                                  ← ② 返回 tool_use（结构化指令，不是文字）
③ 你执行这个工具（查数据库等）
④ 把执行结果发回                 →
                                  ← ⑤ 模型基于结果，返回最终文字回答

Step 6.1：定义工具

// 告诉模型你提供了哪些"能力"
const tools = [
  {
    name: 'get_low_traffic_blogs',
    description: '查询访问量低于指定值的博客文章列表',
    input_schema: {
      type: 'object',
      properties: {
        threshold: { type: 'number', description: '访问量阈值' },
        status:    { type: 'string', enum: ['draft', 'published', 'all'] }
      },
      required: ['threshold']
    }
  },
  {
    name: 'update_blog_title',
    description: '更新指定博客文章的标题',
    input_schema: {
      type: 'object',
      properties: {
        blog_id:   { type: 'string' },
        new_title: { type: 'string' }
      },
      required: ['blog_id', 'new_title']
    }
  }
]

Step 6.2：第一次请求，模型返回工具调用

const res1 = await client.messages.create({
  model: 'claude-sonnet-4-6',
  max_tokens: 1024,
  tools,
  messages: [{
    role: 'user',
    content: '把访问量低于 100 的草稿文章，标题加上 [待优化] 前缀'
  }]
})

console.log(res1.stop_reason)  // 'tool_use'
console.log(res1.content)
// [
//   {
//     type: 'tool_use',
//     id:   'tu_001',
//     name: 'get_low_traffic_blogs',
//     input: { threshold: 100, status: 'draft' }
//   }
// ]

Step 6.3：你执行工具，发回结果

// 根据模型指令执行对应函数
async function executeTool(name: string, input: any) {
  switch (name) {
    case 'get_low_traffic_blogs':
      return await db.query(
        'SELECT id, title, views FROM blogs WHERE views < ? AND status = ?',
        [input.threshold, input.status ?? 'draft']
      )
    case 'update_blog_title':
      await db.query(
        'UPDATE blogs SET title = ? WHERE id = ?',
        [input.new_title, input.blog_id]
      )
      return { success: true, blog_id: input.blog_id }
  }
}

const toolCall = res1.content.find(b => b.type === 'tool_use')
const result   = await executeTool(toolCall.name, toolCall.input)
// result = [{ id: '1', title: '未优化文章', views: 45 }, ...]

// 把结果发回（消息历史必须完整带上）
const res2 = await client.messages.create({
  model: 'claude-sonnet-4-6',
  max_tokens: 1024,
  tools,
  messages: [
    { role: 'user',      content: '把访问量低于 100 的草稿...' },
    { role: 'assistant', content: res1.content },   // 模型上一轮输出
    {
      role: 'user',
      content: [{
        type:        'tool_result',
        tool_use_id: toolCall.id,                   // 必须对应 tu_001
        content:     JSON.stringify(result)
      }]
    }
  ]
})

Step 6.4：模型可能继续调工具

模型拿到文章列表后，会继续调 update_blog_title 逐一更新，直到全部完成，最后返回文字说明。

第一轮：get_low_traffic_blogs → 你查询 → 返回 3 篇文章
第二轮：update_blog_title(blog_id:1) + update_blog_title(blog_id:2) + update_blog_title(blog_id:3)
        ↑ 模型可以一次调用多个工具（并行）
第三轮：end_turn → "已将 3 篇草稿文章标题加上了 [待优化] 前缀"

封装成通用循环（生产代码）

async function runAgent(userMessage: string): Promise<string> {
  const messages: any[] = [{ role: 'user', content: userMessage }]

  while (true) {
    const response = await client.messages.create({
      model: 'claude-sonnet-4-6',
      max_tokens: 1024,
      tools,
      messages
    })

    messages.push({ role: 'assistant', content: response.content })

    if (response.stop_reason === 'end_turn') {
      return response.content.find((b: any) => b.type === 'text').text
    }

    // 并行执行所有工具调用
    const toolResults = await Promise.all(
      response.content
        .filter((b: any) => b.type === 'tool_use')
        .map(async (toolCall: any) => ({
          type:        'tool_result',
          tool_use_id: toolCall.id,
          content:     JSON.stringify(
            await executeTool(toolCall.name, toolCall.input)
          )
        }))
    )

    messages.push({ role: 'user', content: toolResults })
  }
}

---

整体架构图

用户浏览器（React）
      ↕ SSE 流式 / JSON
Next.js API Route（你的后端）
      ↕ 维护 messages 历史
      ↕ 执行工具（查/写 DB）
      ↕ 向量检索
Claude API（模型）

---

落地路径（从今天开始）

今天     →  Step 1-2：写第一个接口，接收文本返回 AI 输出
本周     →  Step 3-4：加多轮对话 + 流式输出，体验质的提升
下周     →  Step 5：把真实数据注入 prompt，让 AI 基于业务数据回答
后续     →  Step 6：加 Tool Use，让 AI 能主动操作数据

---

FAQ

Q：模型真的没有记忆吗？那 ChatGPT 为什么记得我上次说的话？

因为 ChatGPT 的产品层做了历史存储。它在每次对话时，从数据库捞出你的历史消息，拼成 messages[] 发给模型。模型本身仍然是无状态的，"记忆"是产品层实现的。

---

Q：system prompt 和 user message 有什么区别，分开写有什么好处？

system 是"幕后规则"，用户输入的任何内容无法覆盖它（正常情况下）。
user 是每次对话的输入。

分开写的好处：角色设定和约束放 system，不随对话历史增长；用户输入放 messages，保持清晰。如果把 system 混在第一条 user 消息里，每轮对话都要重复发这段文字，浪费 token。

---

Q：temperature 设成 0 不是更好吗？输出最稳定。

不一定。temperature=0 会让模型倾向于选择概率最高的 token，输出死板、重复。
写标题、写文案等创意任务，0.7-0.9 往往比 0 更好用。
只有代码生成、JSON 提取、分类判断等"有唯一正确答案"的任务，才适合调到 0。

---

Q：token 是什么？怎么控制成本？

Token 是模型处理文本的最小单位，粗略理解：

1 token ≈ 0.75 个英文单词 ≈ 0.5 个汉字

计费 = input_tokens × 输入单价 + output_tokens × 输出单价（输出通常贵 3-5 倍）。

控制成本的方法：

1. 选合适的模型（haiku 比 sonnet 便宜约 10 倍）
2. 精简 system prompt，不写废话
3. 限制多轮对话历史长度
4. 生产环境加 prompt cache（重复的 system prompt 只收一次钱）

---

Q：RAG 和 Fine-tuning 怎么选？

RAG：          把数据在查询时注入 prompt
Fine-tuning：  把数据烧进模型权重（改变模型本身）

用 RAG 的情况：
  - 数据经常更新（博客文章、订单数据）
  - 需要引用来源
  - 成本敏感

用 Fine-tuning 的情况：
  - 需要改变模型的输出风格/格式
  - 有大量标注的输入输出对
  - 任务高度专业化

实践结论：90% 的场景 RAG 够用，Fine-tuning 是优化手段，不是入门必须。

---

Q：Tool Use 和直接在代码里查数据库有什么区别？

// 直接查：你的逻辑决定查什么
const data = await db.query('SELECT ...')
const response = await ai.ask(`分析这个数据：${data}`)

// Tool Use：模型的逻辑决定查什么
// 用户说"对比一下最近3个月和去年同期的数据"
// 模型自己推断出要调用 get_stats(period:'3m') 和 get_stats(period:'last_year')
// 你只负责实现 get_stats 函数

本质区别：查询逻辑在哪里。直接查是你写死的，Tool Use 是模型动态决定的。
Tool Use 适合让 AI 处理"用户说的话不固定，需要灵活判断调什么接口"的场景。

---

Q：LangChain / LlamaIndex 这些框架值得学吗？

这些框架帮你封装了：多轮对话历史管理、Tool Use 循环、RAG 流程、向量数据库接入。

什么时候用框架：快速验证想法、不想重复造轮子。
什么时候不用：生产环境需要精细控制、框架版本更新频繁带来不稳定性。

建议：先理解原理，再用框架。本文讲的这些你都懂了，看框架文档就知道它在封装什么，遇到问题才能 debug。反过来，上手框架但不理解底层，一遇到奇怪问题就束手无策。