2.16 应用集成

一句话总结：把 AI 能力嵌入到实际产品中——RAG、Agent、工具调用是三大核心模式。

📊 学习进度
状态：⬜ 未开始
预计时长：6-8 小时
已完成：0/3 个模块
在整体流程中的位置：AI 应用开发·第 4 阶段

📍 本章定位
服务方案：方案 1（核心 80%）/ 方案 2（重要 60%）/ 方案 3（辅助 30%）
学习方式：⭐ 必学
在流程中的作用：这是大多数开发者真正要做的事——把 AI 嵌入产品
核心知识点：RAG、Agent、工具调用、Prompt 管理
预计时长：6-8 小时
完成后能做什么：能搭建一个完整的 RAG 系统或 Agent 应用

人机分工

环节	谁做	重要度	说明
架构设计	🧑 人	⭐⭐⭐⭐⭐	决定用什么模式
代码实现	🤖 AI	⭐⭐⭐⭐	人审核
Prompt 设计	🧑 人 + 🤖 AI	⭐⭐⭐⭐⭐	需要业务理解
用户体验	🧑 人	⭐⭐⭐⭐⭐	AI 无法判断 UX
数据准备	🧑 人 + 🤖 AI	⭐⭐⭐⭐	数据质量决定 RAG 效果
测试验证	🧑 人	⭐⭐⭐⭐	边界场景需人工判断

三大集成模式

模式	适用场景	复杂度	推荐框架
RAG	知识库问答、文档搜索	中	LlamaIndex / LangChain
Agent	复杂工作流、多步骤任务	高	LangGraph / Claude Agent SDK
工具调用	API 集成、数据查询	低	Claude Tool Use / OpenAI Function Calling

模式选择决策矩阵

选择哪种模式？用这个决策矩阵快速判断：

判断维度	选 RAG	选 Agent	选工具调用
核心需求	从大量文档中检索答案	自主完成多步骤任务	调用单个 API 获取数据
数据量	大量非结构化文档	不限	结构化 API
步骤数量	1 步（检索+生成）	3+ 步（有循环/分支）	1 步（调用+返回）
决策复杂度	低（检索即可）	高（需要规划和反思）	低（直接调用）
典型场景	企业知识库问答	自动化工作流	查天气、查汇率
开发成本	中（2-5 天）	高（1-2 周）	低（0.5-1 天）
维护成本	中（需更新文档）	高（需调试逻辑）	低（API 稳定）

混合模式：实际项目中，三种模式经常组合使用。例如 Agent 内部调用 RAG 检索知识，同时使用工具调用获取实时数据。

模式 1：RAG（检索增强生成）

原理：用户问题 → 检索相关文档 → 拼入 Prompt → LLM 生成回答

核心技术栈（2025）：

组件	推荐工具	说明
向量数据库	ChromaDB / Milvus / Qdrant	存储文档向量
Embedding	text-embedding-3-small / BGE-M3	文本转向量
分块策略	语义分块 / 递归分割	文档切片
检索优化	混合检索 / 重排序	提高准确率
框架	LlamaIndex / LangChain	编排整个流程

RAG 完整代码示例（LlamaIndex 实现）

以下是一个生产级 RAG 系统的完整实现，包含文档加载、分块、索引、检索和生成：

python

"""
RAG 系统完整实现 - 基于 LlamaIndex
功能：文档加载 → 分块 → 向量化 → 检索 → 重排序 → 生成
"""
from llama_index.core import (
    VectorStoreIndex,
    SimpleDirectoryReader,
    Settings,
)
from llama_index.core.node_parser import SentenceSplitter
from llama_index.llms.openai import OpenAI
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.core.postprocessor import SentenceTransformerRerank

# ========== 1. 配置模型 ==========
Settings.llm = OpenAI(model="gpt-4o", temperature=0)
Settings.embed_model = OpenAIEmbedding(model="text-embedding-3-small")

# ========== 2. 加载文档 ==========
# 支持 PDF、TXT、Markdown、DOCX 等格式
documents = SimpleDirectoryReader(
    input_dir="./data",
    recursive=True,  # 递归读取子目录
).load_data()

print(f"加载了 {len(documents)} 个文档")

# ========== 3. 配置分块策略 ==========
# 语义分块：按句子边界切分，保留语义完整性
node_parser = SentenceSplitter(
    chunk_size=512,      # 每块最大 token 数
    chunk_overlap=50,    # 块间重叠 token 数（保持上下文连贯）
)
Settings.node_parser = node_parser

# ========== 4. 创建索引 ==========
index = VectorStoreIndex.from_documents(
    documents,
    show_progress=True,
)

# ========== 5. 配置重排序 ==========
# 使用交叉编码器重排序，提高检索精度（据 Cohere 2025 测试，重排序可提升 15-25% 准确率）
reranker = SentenceTransformerRerank(
    top_n=5,  # 重排序后保留前 5 个结果
    model="cross-encoder/ms-marco-MiniLM-L-6-v2",
)

# ========== 6. 创建查询引擎 ==========
query_engine = index.as_query_engine(
    similarity_top_k=20,  # 初步检索 20 个候选
    node_postprocessors=[reranker],  # 重排序
    response_mode="compact",  # 紧凑模式，减少 token 消耗
)

# ========== 7. 查询 ==========
response = query_engine.query("什么是 RAG？它有哪些应用场景？")
print(response)
print("\n参考来源：")
for node in response.source_nodes:
    print(f"  - {node.metadata.get('file_name', '未知')} "
          f"(相似度: {node.score:.3f})")

RAG 优化技巧

RAG 系统的效果 80% 取决于检索质量。以下是经过验证的优化方法：

分块策略对比

策略	原理	优点	缺点	适用场景
固定大小分块	按字符/token 数切分	实现简单	可能切断语义	快速原型
递归分割	按段落→句子→字符逐级切分	保留结构	需调参	通用文档
语义分块	按 embedding 相似度找断点	语义完整	计算成本高	高质量需求
文档结构感知	按标题/章节切分	保留层级	依赖文档格式	Markdown/HTML
命题分块	LLM 提取原子事实	精确度最高	成本最高	精密问答

推荐组合：递归分割（默认）+ 语义分块（高质量场景）+ 重排序兜底。

检索优化方法

python

"""
混合检索：结合向量检索和关键词检索，提高召回率
据 arXiv 2025 研究，混合检索比纯向量检索召回率高 10-20%
"""
from llama_index.core import VectorStoreIndex
from llama_index.retrievers.bm25 import BM25Retriever
from llama_index.core.retrievers import QueryFusionRetriever

# 创建 BM25 关键词检索器
bm25_retriever = BM25Retriever.from_documents(
    documents,
    similarity_top_k=20,
)

# 创建向量检索器
vector_retriever = index.as_retriever(similarity_top_k=20)

# 融合检索：同时使用两种方法，取并集后重排序
fusion_retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    similarity_top_k=10,
    num_queries=4,  # 生成 4 个改写查询
    mode="reciprocal_rerank",
)

# 查询改写示例：用户问"Python 怎么排序"
# 系统自动生成改写：
#   1. "Python 列表排序方法"
#   2. "Python sort 和 sorted 区别"
#   3. "Python 排序算法实现"

重排序的价值

重排序是 RAG 系统中投入产出比最高的优化：

指标	无重排序	有重排序	提升幅度
检索准确率（MRR）	0.62	0.78	+25.8%
回答忠实度（RAGAS）	0.71	0.86	+21.1%
用户满意度	68%	85%	+17pp

数据来源：Cohere Rerank 3.5 基准测试，2025

重排序工作流：向量检索 top-20 → 交叉编码器精排 → 取 top-5 → 送入 LLM。

RAG 演进（2025 趋势）

阶段	特点	适用场景	准确率
Naive RAG	简单检索+生成	快速验证	60-70%
Advanced RAG	查询改写+混合检索+重排序	生产环境	80-90%
Agentic RAG	AI 自主决定查什么、怎么查	复杂场景	90%+

数据来源：LlamaIndex 官方文档，2025

模式 2：Agent（自主执行）

原理：用户意图 → LLM 决策 → 调用工具 → 汇总结果 → 自我反思

关键组件：

组件	说明	推荐工具
规划	任务分解	LangGraph
记忆	短期/长期记忆	LangChain Memory
工具	搜索/代码/API	Tool Use / MCP
反思	自我纠错	Self-Reflection

Agent 完整代码示例（LangGraph 实现）

以下是一个使用 LangGraph 构建的 Agent，支持搜索和代码执行两个工具：

python

"""
Agent 完整实现 - 基于 LangGraph
功能：接收用户问题 → 规划 → 调用工具 → 反思 → 输出结果
"""
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# ========== 1. 定义工具 ==========
@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 实际项目中接入搜索 API（如 Tavily、SerpAPI）
    import requests
    response = requests.get(
        "https://api.tavily.com/search",
        params={"query": query, "max_results": 3},
    )
    return response.json()["results"]

@tool
def run_python(code: str) -> str:
    """执行 Python 代码并返回结果"""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True, text=True, timeout=30,
    )
    return result.stdout or result.stderr

# ========== 2. 配置 LLM ==========
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [search_web, run_python]
llm_with_tools = llm.bind_tools(tools)

# ========== 3. 定义状态 ==========
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# ========== 4. 定义节点 ==========
def agent_node(state: AgentState) -> AgentState:
    """Agent 节点：调用 LLM 决策下一步"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: AgentState) -> str:
    """判断是否需要继续调用工具"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"  # 继续调用工具
    return END  # 结束

# ========== 5. 构建图 ==========
graph = StateGraph(AgentState)

# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))

# 添加边
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("tools", "agent")  # 工具执行后回到 Agent

# 编译
app = graph.compile()

# ========== 6. 运行 ==========
result = app.invoke({
    "messages": [HumanMessage(content="帮我查一下今天比特币的价格，然后用 Python 画个图")]
})
print(result["messages"][-1].content)

Agent 框架对比（2025）

框架	复杂度	多 Agent	状态管理	推荐场景
LangGraph	高	✅	✅ 内置检查点	复杂工作流、生产环境
Claude Agent SDK	中	✅	✅ 原生支持	Claude 生态项目
CrewAI	低	✅	⚠️ 基础	快速原型、团队协作
AutoGen	中	✅	✅	多 Agent 对话

数据来源：GitHub 各项目 Star 数和文档，2025

选择建议：

需要精细控制流程 → LangGraph
用 Claude 模型 → Claude Agent SDK
快速验证想法 → CrewAI
多 Agent 对话场景 → AutoGen

模式 3：工具调用（Function Calling）

原理：LLM 识别意图 → 选择工具 → 生成参数 → 调用 API → 返回结果

工具调用完整代码示例（Claude Tool Use）

python

"""
工具调用完整实现 - 基于 Claude Tool Use
功能：定义工具 → 发送请求 → 处理工具调用 → 返回结果
"""
import anthropic
import json

client = anthropic.Anthropic()

# ========== 1. 定义工具 ==========
tools = [
    {
        "name": "get_weather",
        "description": "获取指定城市的当前天气信息",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "城市名称，如 '北京'、'上海'",
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "温度单位",
                },
            },
            "required": ["city"],
        },
    },
    {
        "name": "get_crypto_price",
        "description": "获取加密货币的当前价格",
        "input_schema": {
            "type": "object",
            "properties": {
                "symbol": {
                    "type": "string",
                    "description": "加密货币符号，如 'BTC'、'ETH'",
                },
            },
            "required": ["symbol"],
        },
    },
]

# ========== 2. 实现工具函数 ==========
def get_weather(city: str, unit: str = "celsius") -> dict:
    """获取天气（示例用模拟数据）"""
    # 实际项目中接入天气 API
    return {"city": city, "temp": 25, "unit": unit, "condition": "晴"}

def get_crypto_price(symbol: str) -> dict:
    """获取加密货币价格（示例用模拟数据）"""
    # 实际项目中接入 CoinGecko 等 API
    prices = {"BTC": 98500, "ETH": 3800}
    return {"symbol": symbol, "price_usd": prices.get(symbol, 0)}

TOOL_MAP = {
    "get_weather": get_weather,
    "get_crypto_price": get_crypto_price,
}

# ========== 3. 发送请求 ==========
def chat_with_tools(user_message: str):
    """带工具调用的对话循环"""
    messages = [{"role": "user", "content": user_message}]
    
    while True:
        # 调用 Claude
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=tools,
            messages=messages,
        )
        
        # 检查是否有工具调用
        if response.stop_reason == "tool_use":
            # 收集所有工具调用结果
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    # 执行工具
                    result = TOOL_MAP[block.name](**block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result, ensure_ascii=False),
                    })
            
            # 将工具结果返回给 Claude
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
        else:
            # 没有工具调用，返回最终回答
            return response.content[0].text

# ========== 4. 运行 ==========
answer = chat_with_tools("今天北京天气怎么样？比特币现在多少钱？")
print(answer)

工具调用实现方式对比

方式	提供商	特点	并行调用	MCP 支持
Tool Use	Anthropic Claude	结构化工具调用	✅	✅
Function Calling	OpenAI	原生支持	✅	⚠️ 实验性
MCP	Anthropic	标准化协议	✅	✅ 原生

API 设计最佳实践

接口设计原则

python

"""
AI API 接口设计最佳实践
"""
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional, List
from enum import Enum

app = FastAPI()

# ========== 1. 定义清晰的数据模型 ==========
class MessageRole(str, Enum):
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"

class Message(BaseModel):
    role: MessageRole
    content: str

class ChatRequest(BaseModel):
    """聊天请求模型"""
    messages: List[Message] = Field(..., description="对话历史")
    model: str = Field(default="claude-sonnet-4-20250514", description="模型名称")
    max_tokens: int = Field(default=1024, ge=1, le=8192, description="最大输出 Token 数")
    temperature: float = Field(default=0.7, ge=0, le=1, description="温度参数")
    stream: bool = Field(default=False, description="是否流式输出")

class ChatResponse(BaseModel):
    """聊天响应模型"""
    content: str = Field(..., description="AI 回答")
    model: str = Field(..., description="使用的模型")
    usage: dict = Field(..., description="Token 使用量")
    latency_ms: float = Field(..., description="响应延迟（毫秒）")

class ErrorResponse(BaseModel):
    """错误响应模型"""
    error: str = Field(..., description="错误类型")
    message: str = Field(..., description="错误信息")
    details: Optional[dict] = Field(None, description="错误详情")

# ========== 2. 实现接口 ==========
@app.post("/v1/chat", response_model=ChatResponse, responses={400: {"model": ErrorResponse}, 500: {"model": ErrorResponse}})
async def chat(request: ChatRequest):
    """
    聊天接口
    
    - **messages**: 对话历史，支持多轮对话
    - **model**: 模型名称，可选 claude-haiku/sonnet/opus
    - **max_tokens**: 最大输出 Token 数
    - **temperature**: 温度参数，0-1
    - **stream**: 是否流式输出
    """
    try:
        # 调用 AI API
        response = await call_ai_api(request)
        return response
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# ========== 3. 健康检查接口 ==========
@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "ok", "timestamp": datetime.now().isoformat()}

# ========== 4. 版本信息接口 ==========
@app.get("/version")
async def version():
    """版本信息"""
    return {"version": "1.0.0", "api_version": "v1"}

错误处理最佳实践

python

"""
AI 应用错误处理最佳实践
"""
import anthropic
from typing import Optional
from enum import Enum

class ErrorCode(str, Enum):
    INVALID_INPUT = "INVALID_INPUT"
    RATE_LIMITED = "RATE_LIMITED"
    MODEL_ERROR = "MODEL_ERROR"
    TIMEOUT = "TIMEOUT"
    INTERNAL_ERROR = "INTERNAL_ERROR"

class AIError(Exception):
    """AI 应用自定义错误"""
    
    def __init__(self, code: ErrorCode, message: str, details: Optional[dict] = None):
        self.code = code
        self.message = message
        self.details = details or {}
        super().__init__(message)

def handle_ai_error(error: Exception) -> dict:
    """统一错误处理"""
    
    if isinstance(error, anthropic.RateLimitError):
        return {
            "code": ErrorCode.RATE_LIMITED,
            "message": "请求过于频繁，请稍后再试",
            "retry_after": 60,
        }
    
    elif isinstance(error, anthropic.APITimeoutError):
        return {
            "code": ErrorCode.TIMEOUT,
            "message": "请求超时，请稍后再试",
            "retry_after": 10,
        }
    
    elif isinstance(error, anthropic.APIError):
        return {
            "code": ErrorCode.MODEL_ERROR,
            "message": f"AI 服务错误: {error.message}",
            "details": {"status_code": error.status_code},
        }
    
    elif isinstance(error, AIError):
        return {
            "code": error.code,
            "message": error.message,
            "details": error.details,
        }
    
    else:
        return {
            "code": ErrorCode.INTERNAL_ERROR,
            "message": "内部错误，请联系管理员",
            "details": {"error": str(error)},
        }

# 使用示例
try:
    response = client.messages.create(...)
except Exception as e:
    error_info = handle_ai_error(e)
    print(f"错误: {error_info['code']} - {error_info['message']}")

MCP 协议详解

MCP（Model Context Protocol）是 Anthropic 在 2024 年底发布的开放标准，定义了 AI 模型与外部工具/数据源的连接方式。可以理解为 AI 的 USB-C 接口——一个通用的工具集成协议 [5]。

MCP 架构

MCP 核心概念

概念	说明	类比
MCP Server	提供工具/数据的服务端	USB 设备
MCP Client	调用工具的 AI 应用	电脑 USB 口
Tool	可调用的函数	设备的功能
Resource	可读取的数据	设备的文件
Prompt	预定义的提示模板	设备的使用说明

MCP 使用示例

python

"""
MCP Server 示例 - 创建一个加密货币价格查询工具
"""
from mcp.server import Server
from mcp.types import Tool, TextContent
import mcp.server.stdio

server = Server("crypto-price-server")

# 定义工具
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_crypto_price",
            description="获取加密货币当前价格",
            inputSchema={
                "type": "object",
                "properties": {
                    "symbol": {
                        "type": "string",
                        "description": "加密货币符号，如 BTC、ETH",
                    },
                },
                "required": ["symbol"],
            },
        ),
    ]

# 实现工具
@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "get_crypto_price":
        symbol = arguments["symbol"]
        # 实际项目中调用 CoinGecko 等 API
        price = 98500 if symbol == "BTC" else 3800
        return [TextContent(
            type="text",
            text=f"{symbol} 当前价格: ${price:,}",
        )]

# 启动服务器
async def main():
    async with mcp.server.stdio.stdio_server() as (read, write):
        await server.run(read, write, server.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

MCP 的价值：一次实现，到处可用。一个 MCP Server 可以被 Claude、Cursor、Windsurf 等任何支持 MCP 的客户端调用。

工程化要点

混合模式架构

实际项目中，三种集成模式经常组合使用：

混合模式代码示例：

python

"""
混合模式：RAG + 工具调用 + Agent
适用于复杂的企业级 AI 应用
"""
from typing import List, Dict
import anthropic

class HybridAIApp:
    """混合模式 AI 应用"""
    
    def __init__(self):
        self.client = anthropic.Anthropic()
        self.rag_engine = None  # RAG 引擎
        self.tools = []         # 工具列表
    
    def process(self, user_input: str, context: Dict = None) -> str:
        """处理用户输入"""
        
        # 1. 意图识别
        intent = self._classify_intent(user_input)
        
        # 2. 根据意图选择处理方式
        if intent == "knowledge_query":
            # RAG 模式：从知识库检索
            return self._rag_process(user_input)
        elif intent == "data_query":
            # 工具调用模式：获取实时数据
            return self._tool_process(user_input)
        elif intent == "complex_task":
            # Agent 模式：多步骤处理
            return self._agent_process(user_input)
        else:
            # 直接生成
            return self._direct_generate(user_input)
    
    def _classify_intent(self, text: str) -> str:
        """意图分类"""
        response = self.client.messages.create(
            model="claude-haiku-4-20250514",  # 用小模型做分类
            max_tokens=100,
            messages=[{
                "role": "user",
                "content": f"""分类以下用户输入的意图：
{text}

返回：knowledge_query / data_query / complex_task / general"""
            }],
        )
        return response.content[0].text.strip()
    
    def _rag_process(self, query: str) -> str:
        """RAG 处理流程"""
        # 1. 检索相关文档
        docs = self.rag_engine.retrieve(query, top_k=5)
        
        # 2. 重排序
        ranked_docs = self.rerank(query, docs)
        
        # 3. 生成回答
        context = "\n".join([d.content for d in ranked_docs[:3]])
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{
                "role": "user",
                "content": f"""根据以下参考资料回答问题。

参考资料：
{context}

问题：{query}

请引用来源。"""
            }],
        )
        return response.content[0].text
    
    def _tool_process(self, query: str) -> str:
        """工具调用处理流程"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            tools=self.tools,
            messages=[{"role": "user", "content": query}],
        )
        
        # 处理工具调用
        if response.stop_reason == "tool_use":
            # 执行工具并返回结果
            tool_results = self._execute_tools(response.content)
            # 继续对话
            response = self.client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=1024,
                messages=[
                    {"role": "user", "content": query},
                    {"role": "assistant", "content": response.content},
                    {"role": "user", "content": tool_results},
                ],
            )
        
        return response.content[0].text
    
    def _agent_process(self, query: str) -> str:
        """Agent 处理流程"""
        # 多步骤处理逻辑
        # ...
        pass
    
    def _direct_generate(self, query: str) -> str:
        """直接生成"""
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": query}],
        )
        return response.content[0].text

Prompt 管理

生产环境中，Prompt 是代码资产，需要版本化管理 [6]：

管理维度	说明	推荐方案
版本控制	每次修改有记录	Git + LangSmith
A/B 测试	对比不同 Prompt 效果	LangSmith Experiments
变量模板	动态插入业务数据	Jinja2 / f-string
质量评估	自动化评估 Prompt 质量	RAGAS / DeepEval

python

"""
Prompt 版本化管理示例
"""
from pathlib import Path
from datetime import datetime

class PromptManager:
    def __init__(self, prompt_dir: str = "./prompts"):
        self.prompt_dir = Path(prompt_dir)
        self.prompt_dir.mkdir(exist_ok=True)
    
    def save(self, name: str, template: str, version: str, metadata: dict = None):
        """保存 Prompt 模板"""
        path = self.prompt_dir / f"{name}_v{version}.txt"
        path.write_text(template, encoding="utf-8")
        
        # 保存元数据
        meta_path = self.prompt_dir / f"{name}_v{version}.meta.json"
        import json
        meta = {
            "name": name,
            "version": version,
            "created_at": datetime.now().isoformat(),
            "metadata": metadata or {},
        }
        meta_path.write_text(json.dumps(meta, indent=2, ensure_ascii=False))
    
    def load(self, name: str, version: str = "latest") -> str:
        """加载 Prompt 模板"""
        if version == "latest":
            files = sorted(self.prompt_dir.glob(f"{name}_v*.txt"))
            if not files:
                raise FileNotFoundError(f"找不到 Prompt: {name}")
            return files[-1].read_text(encoding="utf-8")
        path = self.prompt_dir / f"{name}_v{version}.txt"
        return path.read_text(encoding="utf-8")

# 使用示例
pm = PromptManager()
pm.save("qa_system", "你是{{role}}。请根据以下上下文回答问题：\n{{context}}", "1.0")
pm.save("qa_system", "你是{{role}}。请根据以下上下文回答问题，如果信息不足请说不知道：\n{{context}}", "1.1")

输出解析

LLM 输出是非结构化文本，需要解析为程序可用的数据：

python

"""
结构化输出方案对比
"""
from pydantic import BaseModel
from langchain_core.output_parsers import JsonOutputParser

# 方案 1：JSON Mode（推荐，简单场景）
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=1024,
    messages=[{
        "role": "user",
        "content": "分析这段文本的情感，返回 JSON：{sentiment, confidence, reason}"
    }],
)

# 方案 2：Pydantic 结构化输出（推荐，复杂场景）
class SentimentResult(BaseModel):
    sentiment: str  # positive / negative / neutral
    confidence: float  # 0-1
    reason: str

# 方案 3：LangChain OutputParser
parser = JsonOutputParser(pydantic_object=SentimentResult)

缓存策略实现

生产环境中，缓存是控制成本最有效的手段。据 LangChain 2025 测试，合理缓存可节省 30-50% API 成本。

python

"""
Redis 缓存层 - 精确匹配 + 语义相似匹配
精确匹配：完全相同的问题 → 直接返回缓存
语义相似：意思相近的问题 → 返回相似问题的答案
"""
import hashlib
import json
import redis
from typing import Optional

class AICache:
    """AI 响应缓存"""
    
    def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
        self.redis = redis.from_url(redis_url)
        self.ttl = ttl  # 缓存过期时间（秒）
    
    def _make_key(self, question: str, model: str) -> str:
        """生成缓存键"""
        content = f"{model}:{question.strip().lower()}"
        return f"ai_cache:{hashlib.md5(content.encode()).hexdigest()}"
    
    def get(self, question: str, model: str) -> Optional[str]:
        """查询缓存"""
        key = self._make_key(question, model)
        cached = self.redis.get(key)
        return cached.decode("utf-8") if cached else None
    
    def set(self, question: str, answer: str, model: str):
        """设置缓存"""
        key = self._make_key(question, model)
        self.redis.setex(key, self.ttl, answer)
    
    def invalidate(self, question: str, model: str):
        """手动失效缓存"""
        key = self._make_key(question, model)
        self.redis.delete(key)

# 使用示例
cache = AICache(ttl=7200)  # 2 小时过期

# 查询时先检查缓存
answer = cache.get("什么是 RAG？", "claude-sonnet-4-20250514")
if answer:
    print(f"缓存命中: {answer}")
else:
    answer = call_ai_api("什么是 RAG？")
    cache.set("什么是 RAG？", answer, "claude-sonnet-4-20250514")
    print(f"API 调用: {answer}")

缓存策略选择：

策略	命中率	适用场景	实现复杂度
精确匹配	10-20%	FAQ、固定问题	低
语义相似	30-40%	客服、知识问答	中
分类路由	20-30%	意图分类场景	中
混合策略	40-50%	生产环境	高

流式响应

流式输出是改善用户体验的关键——用户不必等待完整响应：

python

"""
流式响应实现 - SSE (Server-Sent Events)
"""
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import anthropic

app = FastAPI()
client = anthropic.Anthropic()

@app.post("/chat")
async def chat_stream(question: str):
    async def generate():
        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=1024,
            messages=[{"role": "user", "content": question}],
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"
        yield "data: [DONE]\n\n"
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
    )

成本控制

策略	说明	节省幅度
模型路由	简单问题用小模型，复杂问题用大模型	30-50%
缓存	相同/相似问题缓存结果	20-40%
Prompt 精简	减少不必要的指令和示例	10-20%
批处理	合并多个请求	15-25%

python

"""
模型路由示例 - 根据问题复杂度选择模型
"""
def route_model(question: str) -> str:
    """根据问题复杂度路由到不同模型"""
    # 简单判断：问题长度和关键词
    if len(question) < 50 and not any(kw in question for kw in ["分析", "对比", "解释"]):
        return "claude-haiku-4-20250514"  # 快速模型，成本低
    return "claude-sonnet-4-20250514"  # 标准模型

# 使用
model = route_model("今天天气怎么样？")  # → haiku
model = route_model("请对比分析 RAG 和 Fine-tuning 的优缺点")  # → sonnet

OPC 场景下的应用集成 SOP

SOP 1：知识库产品（RAG 场景）

适用场景：企业内部知识库、产品文档问答、客服系统

步骤	时间	负责	产出
1. 收集文档	2-4h	人	结构化文档集
2. 设计分块策略	1-2h	人+AI	分块参数
3. 搭建 RAG Pipeline	2-4h	AI（人审核）	可运行代码
4. 测试问答效果	2-4h	人	测试报告
5. 优化重排序	1-2h	AI（人审核）	优化后配置
6. 部署上线	1-2h	AI	生产环境

总耗时：1-2 天（一个人 + AI 协作）

SOP 2：自动化 Agent（Agent 场景）

适用场景：自动化工作流、数据处理管道、智能助手

步骤	时间	负责	产出
1. 定义任务流程	2-4h	人	流程图
2. 设计工具接口	1-2h	人	工具定义文档
3. 实现 Agent	4-8h	AI（人审核）	可运行代码
4. 调试边界场景	4-8h	人	测试用例
5. 添加错误处理	2-4h	AI	健壮代码
6. 部署监控	2-4h	AI	生产环境

总耗时：3-5 天（一个人 + AI 协作）

SOP 3：API 集成（工具调用场景）

适用场景：对接外部 API、数据查询、简单自动化

步骤	时间	负责	产出
1. 梳理 API 需求	1-2h	人	API 清单
2. 定义工具 Schema	1h	人+AI	工具定义
3. 实现调用逻辑	2-4h	AI（人审核）	可运行代码
4. 测试 + 部署	1-2h	人	生产环境

总耗时：0.5-1 天（一个人 + AI 协作）

常见问题

问题	原因	解决方案
RAG 回答不准	检索质量低	优化分块 + 混合检索 + 重排序
RAG 回答幻觉	LLM 编造信息	添加引用来源 + 限制生成
Agent 死循环	缺少退出条件	设置最大迭代次数 + 超时
Agent 选错工具	工具描述不清	优化工具 description
成本失控	Token 消耗过多	模型路由 + 缓存 + Prompt 精简
响应太慢	检索或生成耗时长	流式输出 + 异步处理
工具调用失败	参数格式错误	Pydantic 校验 + 重试机制
向量数据库查询慢	索引未优化	使用 HNSW 索引 + 减少 top_k
RAG 上下文窗口溢出	检索结果太多	限制检索数量 + 压缩上下文
多轮对话丢失上下文	未管理对话历史	滑动窗口 + 摘要压缩

下一步

完成应用集成后，进入阶段 5：部署上线

参考与延伸

[1] LangChain. "Documentation"（2025）— RAG/Agent 框架，覆盖从简单链到复杂 Agent 的完整工具链

[2] LlamaIndex. "Documentation"（2025）— RAG 专用框架，支持多种数据源和索引策略

[3] Anthropic. "Model Context Protocol"（2025）— MCP 协议文档，定义 AI 与外部工具的标准化连接方式

[4] LangGraph. "Documentation"（2025）— Agent 编排框架，支持有状态的图执行和检查点

[5] Cohere. "Rerank Documentation"（2025）— 重排序技术文档，交叉编码器提升检索精度 15-25%

[6] Anthropic. "Tool Use Documentation"（2025）— Claude 工具调用 API，支持并行调用和 MCP 集成

[7] RAGAS. "RAG Evaluation Framework"（2025）— RAG 系统评估框架，量化忠实度、相关性等指标

[8] FastAPI. "Documentation"（2025）— Python Web 框架，适合 AI API 开发

[9] Pydantic. "Documentation"（2025）— 数据验证库，用于 API 接口设计

[10] Qdrant. "Vector Database Documentation"（2025）— 高性能向量数据库，支持过滤和多租户

[11] Anthropic. "Prompt Caching"（2025）— Prompt 缓存功能，重复前缀自动缓存，降低 90% 输入成本

2.16 应用集成 ​

人机分工 ​

三大集成模式 ​

模式选择决策矩阵 ​

模式 1：RAG（检索增强生成） ​

RAG 完整代码示例（LlamaIndex 实现） ​

RAG 优化技巧 ​

分块策略对比 ​

检索优化方法 ​

重排序的价值 ​

RAG 演进（2025 趋势） ​

模式 2：Agent（自主执行） ​

Agent 完整代码示例（LangGraph 实现） ​

Agent 框架对比（2025） ​

模式 3：工具调用（Function Calling） ​

工具调用完整代码示例（Claude Tool Use） ​

工具调用实现方式对比 ​

API 设计最佳实践 ​

接口设计原则 ​

错误处理最佳实践 ​

MCP 协议详解 ​

MCP 架构 ​

MCP 核心概念 ​

MCP 使用示例 ​

工程化要点 ​

混合模式架构 ​

Prompt 管理 ​

输出解析 ​

缓存策略实现 ​

流式响应 ​

成本控制 ​

OPC 场景下的应用集成 SOP ​

SOP 1：知识库产品（RAG 场景） ​

SOP 2：自动化 Agent（Agent 场景） ​

SOP 3：API 集成（工具调用场景） ​

常见问题 ​

下一步 ​

参考与延伸 ​

2.16 应用集成

人机分工

三大集成模式

模式选择决策矩阵

模式 1：RAG（检索增强生成）

RAG 完整代码示例（LlamaIndex 实现）

RAG 优化技巧

分块策略对比

检索优化方法

重排序的价值

RAG 演进（2025 趋势）

模式 2：Agent（自主执行）

Agent 完整代码示例（LangGraph 实现）

Agent 框架对比（2025）

模式 3：工具调用（Function Calling）

工具调用完整代码示例（Claude Tool Use）

工具调用实现方式对比

API 设计最佳实践

接口设计原则

错误处理最佳实践

MCP 协议详解

MCP 架构

MCP 核心概念

MCP 使用示例

工程化要点

混合模式架构

Prompt 管理

输出解析

缓存策略实现

流式响应

成本控制

OPC 场景下的应用集成 SOP

SOP 1：知识库产品（RAG 场景）

SOP 2：自动化 Agent（Agent 场景）

SOP 3：API 集成（工具调用场景）

常见问题

下一步

参考与延伸