AI-原生开发
概览
AI-原生开发专注于构建将AI作为一等公民的应用，而不是事后的想法。这项技能提供了全面的模式，用于集成大型语言模型（LLM）、实现检索增强生成（RAG）、使用向量数据库、构建智能工作流以及优化AI应用性能和成本。
何时使用这项技能：
- 构建聊天机器人、问答系统或对话界面
- 实施语义搜索或推荐引擎
- 创建可以使用工具并采取行动的AI代理
- 将LLM（OpenAI、Anthropic、开源模型）集成到应用中
- 构建用于知识检索的RAG系统
- 优化AI成本和延迟
- 实施AI可观测性和监控
为什么AI-原生开发很重要
传统软件是确定性的；AI-原生应用是概率性的：
- 上下文就是一切：LLM需要相关上下文才能提供准确的答案
- RAG优于微调：检索比微调更便宜、更灵活
- 嵌入启用语义搜索：超越关键词匹配，理解含义
- 智能工作流：LLM可以自主地推理、规划和使用工具
- 成本管理：令牌使用直接影响运营成本
- 可观测性：调试概率系统需要新方法
- 提示工程：你提问的方式和提问的内容一样重要
核心概念
1. 嵌入和向量搜索
嵌入是捕获语义含义的文本向量表示。相似的概念具有相似的向量。
关键能力：
- 将文本转换为高维向量（1536或3072维）
- 使用余弦相似度测量语义相似性
- 通过向量搜索找到相关文档
- 批量处理以提高效率
详细实现： 见 references/vector-databases.md 文件：
- OpenAI嵌入设置和批量处理
- 余弦相似度算法
- 分块策略（500-1000个令牌，重叠10-20%）
2. 向量数据库
在规模上高效存储和检索嵌入。
流行选项：
- Pinecone：无服务器，托管服务（每小时0.096美元）
- Chroma：开源，自托管
- Weaviate：灵活的模式，混合搜索
- Qdrant：基于Rust，高性能
详细实现： 见 references/vector-databases.md 文件：
- 每个数据库的完整设置指南
- 插入、查询、更新、删除操作
- 元数据过滤和混合搜索
- 成本比较和最佳实践
3. RAG（检索增强生成）
RAG将检索系统与LLM结合起来，提供准确、有根据的答案。
核心模式：
- 从向量数据库中检索相关文档
- 从顶部结果构建上下文
- 使用检索到的上下文和LLM生成答案
高级模式：
- 带引用和源跟踪的RAG
- 混合搜索（语义+关键词）
- 多查询RAG以获得更好的召回率
- HyDE（假设文档嵌入）
- 用于相关性的上下文压缩
详细实现： 见 references/rag-patterns.md 文件：
- 基本和高级RAG模式，带有完整代码
- 引用策略
- 带有互惠排名融合的混合搜索
- 对话记忆模式
- 错误处理和验证
4. 函数调用和工具使用
使LLM能够可靠地使用外部工具和API。
能力：
- 使用JSON模式定义工具
- 根据LLM决策执行函数
- 处理并行工具调用
- 使用工具时流式响应
详细实现： 见 references/function-calling.md 文件：
- 工具定义模式（OpenAI和Anthropic）
- 函数调用循环
- 并行和流式工具执行
- 使用Zod进行输入验证
- 错误处理和回退策略
5. 智能工作流
使LLM能够推理、规划和自主采取行动。
模式：
- ReAct：带有观察的推理+行动循环
- 思维树：探索多个推理路径
- 多代理：在复杂任务上协作的专门代理
- 自主代理：自我指导的目标实现
详细实现： 见 references/agentic-workflows.md 文件：
- 完整的ReAct循环实现
- 思维树探索
- 多代理协调模式
- 代理内存管理
- 错误恢复和安全保护
5.1 多代理协调（Opus 4.5）
利用Opus 4.5的扩展思考能力，使用高级多代理模式。
何时使用扩展思考：
- 协调3+专门代理
- 代理输出之间的复杂依赖关系解析
- 基于代理能力的动态任务分配
- 当代理产生矛盾结果时的冲突解决
协调器模式：
interface AgentTask {
id: string;
type: 'research' | 'code' | 'review' | 'design';
input: unknown;
dependencies: string[]; // 必须首先完成的任务ID
}
interface AgentResult {
taskId: string;
output: unknown;
confidence: number;
reasoning: string;
}
async function orchestrateAgents(
goal: string,
availableAgents: Agent[]
): Promise<AgentResult[]> {
// 第1步：使用扩展思考将目标分解为任务
const taskPlan = await planTasks(goal, availableAgents);
// 第2步：构建依赖图
const dependencyGraph = buildDependencyGraph(taskPlan.tasks);
// 第3步：执行任务时尊重依赖关系
const results: AgentResult[] = [];
const completed = new Set<string>();
while (completed.size < taskPlan.tasks.length) {
// 查找满足依赖关系的任务
const ready = taskPlan.tasks.filter(task =>
!completed.has(task.id) &&
task.dependencies.every(dep => completed.has(dep))
);
// 并行执行准备就绪的任务
const batchResults = await Promise.all(
ready.map(task => executeAgentTask(task, availableAgents))
);
// 验证结果 - 使用扩展思考解决冲突
const validatedResults = await validateAndResolveConflicts(
batchResults,
results
);
results.push(...validatedResults);
ready.forEach(task => completed.add(task.id));
}
return results;
}
带有扩展思考的任务规划：
基于 Anthropic的扩展思考文档：
import Anthropic from '@anthropic-ai/sdk';
const anthropic = new Anthropic();
async function planTasks(
goal: string,
agents: Agent[]
): Promise<{ tasks: AgentTask[]; rationale: string }> {
// 扩展思考需要budget_tokens < max_tokens
// 最小预算：1,024令牌
const response = await anthropic.messages.create({
model: 'claude-opus-4-5-20251101', // 或 claude-sonnet-4-5-20250929
max_tokens: 16000,
thinking: {
type: 'enabled',
budget_tokens: 10000 // 扩展思考用于复杂规划
},
messages: [{
role: 'user',
content: `
目标：${goal}
可用代理及其能力：
${agents.map(a => `- ${a.name}: ${a.capabilities.join(', ')}`).join('
')}
将此目标分解为任务。对于每个任务，请指定：
1. 哪个代理应处理它
2. 它需要什么输入
3. 它依赖于哪些其他任务
4. 预期的输出格式
仔细考虑：
- 最佳并行化机会
- 代理输出之间的潜在冲突
- 需要在任务之间流动的信息
`
}]
});
// 响应包含思考块，后跟文本块
// content: [{ type: 'thinking', thinking: '...' }, { type: 'text', text: '...' }]
return parseTaskPlan(response);
}
冲突解决：
async function validateAndResolveConflicts(
newResults: AgentResult[],
existingResults: AgentResult[]
): Promise<AgentResult[]> {
// 检查与现有结果的冲突
const conflicts = detectConflicts(newResults, existingResults);
if (conflicts.length === 0) {
return newResults;
}
// 使用扩展思考解决冲突
const resolution = await anthropic.messages.create({
model: 'claude-opus-4-5-20251101',
max_tokens: 8000,
thinking: {
type: 'enabled',
budget_tokens: 5000
},
messages: [{
role: 'user',
content: `
以下代理输出存在冲突：
${conflicts.map(c => `
冲突：${c.description}
代理A (${c.agentA.name}): ${JSON.stringify(c.resultA)}
代理B (${c.agentB.name}): ${JSON.stringify(c.resultB)}
`).join('
')}
分析每个冲突并确定：
1. 哪个输出更有可能是正确的，为什么
2. 如果两者都有价值，如何综合它们
3. 需要哪些额外的验证
`
}]
});
return applyResolutions(newResults, resolution);
}
自适应代理选择：
async function selectOptimalAgent(
task: AgentTask,
agents: Agent[],
context: ExecutionContext
): Promise<Agent> {
// 根据以下因素为每个代理打分：
// - 能力匹配
// - 当前负载
// - 在类似任务上的历史表现
// - 成本（模型层级）
const scores = agents.map(agent => ({
agent,
score: calculateAgentScore(agent, task, context)
}));
// 对于复杂任务，使用Opus；对于简单任务，使用Haiku
const complexity = assessTaskComplexity(task);
if (complexity > 0.7) {
// 过滤可以使用Opus的代理
const opusCapable = scores.filter(s => s.agent.supportsOpus);
return opusCapable.sort((a, b) => b.score - a.score)[0].agent;
}
return scores.sort((a, b) => b.score - a.score)[0].agent;
}
代理通信协议：
interface AgentMessage {
from: string;
to: string | 'broadcast';
type: 'request' | 'response' | 'update' | 'conflict';
payload: unknown;
timestamp: Date;
}
class AgentCommunicationBus {
private messages: AgentMessage[] = [];
private subscribers: Map<string, (msg: AgentMessage) => void> = new Map();
send(message: AgentMessage): void {
this.messages.push(message);
if (message.to === 'broadcast') {
this.subscribers.forEach(callback => callback(message));
} else {
this.subscribers.get(message.to)?.(message);
}
}
subscribe(agentId: string, callback: (msg: AgentMessage) => void): void {
this.subscribers.set(agentId, callback);
}
getHistory(agentId: string): AgentMessage[] {
return this.messages.filter(
m => m.from === agentId || m.to === agentId || m.to === 'broadcast'
);
}
}
6. 流式响应
提供实时AI响应以改善用户体验。
能力：
- 逐令牌流式传输LLM输出
- 服务器发送事件（SSE）适用于Web客户端
- 流式函数调用
- 后压处理
详细实现： 见 ../streaming-api-patterns/SKILL.md 文件中的流式模式
7. 成本优化
策略：
- 对于简单任务使用较小的模型（GPT-3.5 vs GPT-4）
- 实施提示缓存（Anthropic的临时缓存）
- 尽可能批量请求
- 设置max_tokens以防止生成失控
- 使用警报监控使用情况
令牌计数：
import { encoding_for_model } from 'tiktoken'
function countTokens(text: string, model = 'gpt-4'): number {
const encoder = encoding_for_model(model)
const tokens = encoder.encode(text)
encoder.free()
return tokens.length
}
详细实现： 见 references/observability.md 文件：
- 成本估算和预算跟踪
- 模型选择策略
- 提示缓存模式
8. 可观测性与监控
在生产中跟踪LLM性能、成本和质量。
工具：
- LangSmith：跟踪、评估、监控
- LangFuse：开源可观测性
- 自定义日志记录：带有指标的结构化日志
关键指标：
- 吞吐量（每分钟请求数）
- 延迟（P50、P95、P99）
- 令牌使用和成本
- 错误率
- 质量分数（相关性、连贯性、事实性）
详细实现： 见 references/observability.md 文件：
- LangSmith和LangFuse集成
- 自定义日志记录器实现
- 性能监控
- 质量评估
- 调试和错误分析
搜索参考
这项技能包括详细的参考资料。使用grep查找特定模式：
# 查找RAG模式
grep -r "RAG" references/
# 搜索特定的向量数据库
grep -A 10 "Pinecone Setup" references/vector-databases.md
# 查找智能工作流示例
grep -B 5 "ReAct Pattern" references/agentic-workflows.md
# 定位函数调用模式
grep -n "parallel.*tool" references/function-calling.md
# 搜索成本优化
grep -i "cost\|pricing\|budget" references/observability.md
# 查找所有嵌入的代码示例
grep -A 20 "async function.*embedding" references/
最佳实践
上下文管理
- ✅ 保持上下文窗口在模型限制的75%以下
- ✅ 对长对话使用滑动窗口
- ✅ 在滚动出之前总结旧消息
- ✅ 移除冗余或不相关的上下文
嵌入策略
- ✅ 将文档分块为500-1000个令牌
- ✅ 通过10-20%的重叠来保持连续性
- ✅ 与分块一起包含元数据（标题、来源、日期）
- ✅ 当源数据更改时重新嵌入
RAG质量
- ✅ 使用混合搜索（语义+关键词）
- ✅ 为相关性重新排列结果
- ✅ 在上下文中包含引用/来源
- ✅ 为事实答案设置低温（0.1-0.3）
- ✅ 根据检索到的上下文验证答案
函数调用
- ✅ 提供清晰、简洁的函数描述
- ✅ 使用严格的JSON模式参数
- ✅ 优雅地处理缺失或无效参数
- ✅ 限制在10-20个工具以避免混淆
- ✅ 在返回给LLM之前验证函数输出
成本优化
- ✅ 对于简单任务使用较小的模型
- ✅ 实施提示缓存以重复内容
- ✅ 尽可能批量请求
- ✅ 设置max_tokens以防止生成失控
- ✅ 使用警报监控异常使用情况
安全
- ✅ 验证和清理用户输入
- ✅ 不要在提示中包含机密信息
- ✅ 实施速率限制
- ✅ 过滤有害内容的输出
- ✅ 每个环境使用单独的API密钥
模板
使用提供的模板进行常见的AI模式：
templates/rag-pipeline.ts- 基本RAG实现templates/agentic-workflow.ts- ReAct代理模式
示例
完整的RAG聊天机器人
见 examples/chatbot-with-rag/ 文件夹，了解全栈实现：
- 向量数据库设置与文档摄取
- 带引用的RAG查询
- 流式聊天界面
- 成本跟踪和监控
检查表
AI实施检查表
见 checklists/ai-implementation.md 文件，进行全面验证，涵盖：
- [ ] 向量数据库设置和配置
- [ ] 嵌入生成和分块策略
- [ ] 带质量验证的RAG管道
- [ ] 函数调用与错误处理
- [ ] 流式响应实现
- [ ] 成本监控和预算警报
- [ ] 可观测性和日志记录
- [ ] 安全和输入验证
常见模式
语义缓存
通过缓存类似查询来降低成本：
const cache = new Map<string, { embedding: number[]; response: string }>()
async function cachedRAG(query: string) {
const queryEmbedding = await createEmbedding(query)
// 检查缓存中是否存在类似查询
for (const [cachedQuery, cached] of cache.entries()) {
const similarity = cosineSimilarity(queryEmbedding, cached.embedding)
if (similarity > 0.95) {
return cached.response
}
}
// 未缓存，执行RAG
const response = await ragQuery(query)
cache.set(query, { embedding: queryEmbedding, response })
return response
}
对话记忆
在多个回合中保持上下文：
interface ConversationMemory {
messages: Message[] // 最后10条消息
summary?: string // 旧消息的摘要
}
async function getConversationContext(userId: string): Promise<Message[]> {
const memory = await db.memory.findUnique({ where: { userId } })
return [
{ role: 'system', content: `之前对话的摘要：${memory.summary}` },
...memory.messages.slice(-5) // 最后5条消息
]
}
提示工程
少数样本学习
提供示例以指导LLM行为：
const fewShotExamples = `
示例1：
输入："I love this product!"
情感：积极
示例2：
输入："It's okay, nothing special"
情感：中立
`
// 包含在系统提示中
思维链（CoT）
要求LLM显示推理：
const prompt = `${problem}
让我们逐步思考：`
资源
下一步
掌握AI-原生开发后：
- 探索流式API模式技能，以获得实时AI响应
- 使用类型安全和验证技能进行AI输入/输出验证
- 应用边缘计算模式技能，以实现全球AI部署
- 参考可观测性模式以进行生产监控