自己接模型开发 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 倍）。
控制成本的方法：
- 选合适的模型（haiku 比 sonnet 便宜约 10 倍）
- 精简 system prompt，不写废话
- 限制多轮对话历史长度
- 生产环境加 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。反过来，上手框架但不理解底层，一遇到奇怪问题就束手无策。
登录
