AI应用架构模式

架构视角：企业AI应用的核心挑战

随着大语言模型（LLM）技术的成熟，企业级AI应用正从概念验证走向生产部署。然而，从"玩具级"演示到"生产级"应用之间存在巨大的工程鸿沟。本文从架构师视角，系统梳理AI应用的核心架构模式与工程实践。

RAG模式：检索增强生成

RAG（Retrieval-Augmented Generation）是当前最主流的AI应用架构，通过将外部知识库与LLM结合，有效缓解幻觉问题并注入领域知识。

RAG核心架构

# RAG系统核心组件
from dataclasses import dataclass
from typing import List, Optional
import numpy as np

@dataclass
class Document:
    """文档块数据结构"""
    id: str
    content: str
    metadata: dict
    embedding: Optional[np.ndarray] = None

class RAGPipeline:
    """
    RAG流水线：检索 + 生成
    遵循关注点分离原则，每个组件可独立测试和替换
    """
    
    def __init__(
        self,
        embedding_model: EmbeddingModel,
        vector_store: VectorStore,
        llm: LLMInterface,
        reranker: Optional[Reranker] = None
    ):
        self.embedding_model = embedding_model
        self.vector_store = vector_store
        self.llm = llm
        self.reranker = reranker
    
    async def query(self, question: str, top_k: int = 5) -> str:
        """
        完整的RAG查询流程
        """
        # 1. 查询向量化
        query_embedding = await self.embedding_model.encode(question)
        
        # 2. 向量检索（召回阶段）
        candidates = await self.vector_store.similarity_search(
            query_embedding, 
            k=top_k * 2  # 多召回一些供重排序
        )
        
        # 3. 重排序优化（可选）
        if self.reranker:
            candidates = await self.reranker.rerank(question, candidates, k=top_k)
        
        # 4. 构建增强提示
        context = self._build_context(candidates[:top_k])
        prompt = self._build_prompt(question, context)
        
        # 5. LLM生成
        answer = await self.llm.generate(prompt)
        
        return answer
    
    def _build_context(self, documents: List[Document]) -> str:
        """构建上下文，包含来源标记便于追溯"""
        contexts = []
        for i, doc in enumerate(documents, 1):
            contexts.append(f"[文档{i}] {doc.content}\n来源: {doc.metadata.get('source', '未知')}")
        return "\n\n".join(contexts)
    
    def _build_prompt(self, question: str, context: str) -> str:
        """构建结构化提示模板"""
        return f"""基于以下参考文档回答问题。如果文档中没有相关信息，请明确说明。

参考文档：
{context}

问题：{question}

请提供准确、简洁的回答，并注明信息来源："""

文档处理与索引

class DocumentProcessor:
    """
    文档处理流水线：解析 -> 清洗 -> 分块 -> 向量化
    """
    
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
    
    async def process_document(self, file_path: str) -> List[Document]:
        # 1. 文档解析
        raw_text = await self._parse_document(file_path)
        
        # 2. 文本清洗
        cleaned_text = self._clean_text(raw_text)
        
        # 3. 智能分块
        chunks = self._chunk_text(cleaned_text)
        
        # 4. 生成文档对象
        documents = [
            Document(
                id=f"{file_path}_{i}",
                content=chunk,
                metadata={"source": file_path, "chunk_index": i}
            )
            for i, chunk in enumerate(chunks)
        ]
        
        return documents
    
    def _chunk_text(self, text: str) -> List[str]:
        """
        递归字符分块策略
        优先按段落分割，段落过长则按句子，句子过长则按字符
        """
        separators = ["\n\n", "\n", "。", "！", "？", " ", ""]
        chunks = []
        
        for separator in separators:
            if not text:
                break
            
            parts = text.split(separator)
            current_chunk = ""
            
            for part in parts:
                if len(current_chunk) + len(part) < self.chunk_size:
                    current_chunk += part + separator
                else:
                    if current_chunk:
                        chunks.append(current_chunk.strip())
                    current_chunk = part + separator
            
            if current_chunk:
                chunks.append(current_chunk.strip())
            
            # 如果分块效果良好，停止尝试更细粒度的分隔符
            if all(len(chunk) <= self.chunk_size * 1.2 for chunk in chunks):
                break
            else:
                # 重新尝试更细粒度
                text = separator.join(chunks)
                chunks = []
        
        return chunks or [text]

Agent模式：智能体工作流

AI Agent是大模型应用的下一个前沿，具备规划、记忆、工具使用等能力，能够自主完成复杂任务。

ReAct架构实现

from enum import Enum
from typing import List, Dict, Any, Optional
import json

class ActionType(Enum):
    THOUGHT = "thought"
    ACTION = "action"
    OBSERVATION = "observation"
    FINISH = "finish"

@dataclass
class Step:
    type: ActionType
    content: str
    tool_name: Optional[str] = None
    tool_input: Optional[Dict] = None

class ReActAgent:
    """
    ReAct (Reasoning + Acting) 智能体
    通过思考-行动-观察的循环解决复杂问题
    """
    
    def __init__(
        self,
        llm: LLMInterface,
        tools: Dict[str, Tool],
        max_iterations: int = 10
    ):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations
    
    async def run(self, query: str) -> str:
        """
        执行ReAct循环
        """
        context = []
        
        for iteration in range(self.max_iterations):
            # 1. 思考下一步
            thought = await self._think(query, context)
            context.append(Step(ActionType.THOUGHT, thought))
            
            # 2. 决定行动
            action = await self._decide_action(query, context)
            
            if action.type == ActionType.FINISH:
                return action.content
            
            context.append(action)
            
            # 3. 执行工具
            observation = await self._execute_tool(action)
            context.append(Step(ActionType.OBSERVATION, observation))
            
            # 4. 检查是否陷入循环
            if self._is_looping(context):
                return "思考过程陷入循环，请简化问题重试"
        
        return "达到最大迭代次数，未能完成目标"
    
    async def _think(self, query: str, context: List[Step]) -> str:
        """生成思考过程"""
        prompt = self._build_react_prompt(query, context)
        response = await self.llm.generate(prompt)
        return response
    
    async def _decide_action(self, query: str, context: List[Step]) -> Step:
        """解析LLM输出，决定下一步行动"""
        # 实现解析逻辑...
        pass
    
    async def _execute_tool(self, action: Step) -> str:
        """执行工具调用"""
        if action.tool_name not in self.tools:
            return f"错误：未知工具 {action.tool_name}"
        
        tool = self.tools[action.tool_name]
        try:
            result = await tool.run(**action.tool_input)
            return json.dumps(result, ensure_ascii=False)
        except Exception as e:
            return f"工具执行错误: {str(e)}"
    
    def _is_looping(self, context: List[Step]) -> bool:
        """检测是否陷入循环"""
        if len(context) < 6:
            return False
        
        # 检查最近几步是否重复
        recent_thoughts = [s.content for s in context[-6:] if s.type == ActionType.THOUGHT]
        return len(set(recent_thoughts)) < len(recent_thoughts) / 2

# 工具定义示例
class SearchTool(Tool):
    """搜索工具"""
    name = "search"
    description = "使用搜索引擎查询信息"
    
    async def run(self, query: str) -> Dict[str, Any]:
        # 实现搜索逻辑
        return {"results": [...]}

class CalculatorTool(Tool):
    """计算器工具"""
    name = "calculator"
    description = "执行数学计算"
    
    async def run(self, expression: str) -> Dict[str, Any]:
        try:
            result = eval(expression)  # 实际使用应更安全的方式
            return {"result": result}
        except Exception as e:
            return {"error": str(e)}

多Agent协作架构

class MultiAgentSystem:
    """
    多Agent协作系统
    不同Agent负责不同领域，通过消息总线协作
    """
    
    def __init__(self):
        self.agents: Dict[str, BaseAgent] = {}
        self.message_bus = MessageBus()
    
    def register_agent(self, agent: BaseAgent):
        self.agents[agent.name] = agent
        agent.set_message_bus(self.message_bus)
    
    async def delegate_task(self, task: Task) -> TaskResult:
        """
        任务委派与协调
        """
        # 1. 任务分析
        planner = self.agents.get("planner")
        subtasks = await planner.plan(task)
        
        # 2. 并行执行
        results = await asyncio.gather(*[
            self._execute_subtask(subtask)
            for subtask in subtasks
        ])
        
        # 3. 结果整合
        integrator = self.agents.get("integrator")
        final_result = await integrator.integrate(results)
        
        return final_result
    
    async def _execute_subtask(self, subtask: SubTask) -> SubTaskResult:
        """执行子任务"""
        agent = self.agents.get(subtask.required_capability)
        if not agent:
            raise ValueError(f"没有具备 {subtask.required_capability} 能力的Agent")
        
        return await agent.execute(subtask)

微调模式：领域适应优化

何时需要微调？

LoRA微调实战

from peft import LoraConfig, get_peft_model, TaskType
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from trl import SFTTrainer

class LoRAFineTuner:
    """
    LoRA (Low-Rank Adaptation) 高效微调
    只训练低秩矩阵，大幅降低显存需求
    """
    
    def __init__(
        self,
        base_model_name: str,
        output_dir: str,
        r: int = 16,  # LoRA秩
        lora_alpha: int = 32,
        lora_dropout: float = 0.05
    ):
        self.base_model_name = base_model_name
        self.output_dir = output_dir
        
        # LoRA配置
        self.lora_config = LoraConfig(
            r=r,
            lora_alpha=lora_alpha,
            target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
            lora_dropout=lora_dropout,
            bias="none",
            task_type=TaskType.CAUSAL_LM
        )
    
    def prepare_model(self):
        """加载并配置模型"""
        # 4-bit量化加载，节省显存
        model = AutoModelForCausalLM.from_pretrained(
            self.base_model_name,
            load_in_4bit=True,
            torch_dtype=torch.float16,
            device_map="auto"
        )
        
        # 应用LoRA
        model = get_peft_model(model, self.lora_config)
        model.print_trainable_parameters()
        
        return model
    
    def train(
        self,
        train_dataset,
        eval_dataset=None,
        num_epochs: int = 3,
        batch_size: int = 4,
        learning_rate: float = 2e-4
    ):
        """执行训练"""
        model = self.prepare_model()
        tokenizer = AutoTokenizer.from_pretrained(self.base_model_name)
        
        training_args = TrainingArguments(
            output_dir=self.output_dir,
            num_train_epochs=num_epochs,
            per_device_train_batch_size=batch_size,
            learning_rate=learning_rate,
            logging_steps=10,
            save_strategy="epoch",
            evaluation_strategy="epoch" if eval_dataset else "no",
            load_best_model_at_end=True,
            fp16=True,
            gradient_accumulation_steps=4,
            optim="paged_adamw_8bit"  # 分页优化器节省显存
        )
        
        trainer = SFTTrainer(
            model=model,
            tokenizer=tokenizer,
            args=training_args,
            train_dataset=train_dataset,
            eval_dataset=eval_dataset,
            dataset_text_field="text",
            max_seq_length=512
        )
        
        trainer.train()
        
        # 保存LoRA权重
        model.save_pretrained(f"{self.output_dir}/lora_adapter")
        tokenizer.save_pretrained(self.output_dir)

LLMOps：生产环境架构

模型服务架构

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import asyncio
from typing import AsyncGenerator
import time

app = FastAPI()

class LLMService:
    """
    生产级LLM服务
    包含限流、缓存、监控等企业级特性
    """
    
    def __init__(self):
        self.model = None
        self.tokenizer = None
        self.cache = RedisCache()
        self.rate_limiter = TokenBucketRateLimiter(
            capacity=1000,  # 每分钟最大请求数
            refill_rate=1000/60
        )
        self.metrics = MetricsCollector()
    
    async def generate(
        self,
        prompt: str,
        max_tokens: int = 512,
        temperature: float = 0.7,
        stream: bool = False
    ) -> AsyncGenerator[str, None]:
        """
        文本生成接口
        """
        # 1. 限流检查
        if not await self.rate_limiter.acquire():
            raise HTTPException(429, "请求过于频繁")
        
        # 2. 缓存检查
        cache_key = self._generate_cache_key(prompt, max_tokens, temperature)
        cached = await self.cache.get(cache_key)
        if cached:
            self.metrics.record_cache_hit()
            yield cached
            return
        
        # 3. 执行生成
        start_time = time.time()
        
        try:
            if stream:
                async for token in self._stream_generate(prompt, max_tokens, temperature):
                    yield token
            else:
                response = await self._batch_generate(prompt, max_tokens, temperature)
                # 缓存结果
                await self.cache.set(cache_key, response, ttl=3600)
                yield response
            
            # 4. 记录指标
            latency = time.time() - start_time
            self.metrics.record_request(latency=latency)
            
        except Exception as e:
            self.metrics.record_error(type(e).__name__)
            raise
    
    async def _stream_generate(
        self,
        prompt: str,
        max_tokens: int,
        temperature: float
    ) -> AsyncGenerator[str, None]:
        """流式生成"""
        inputs = self.tokenizer(prompt, return_tensors="pt")
        
        streamer = TextIteratorStreamer(self.tokenizer)
        
        generation_kwargs = dict(
            inputs,
            streamer=streamer,
            max_new_tokens=max_tokens,
            temperature=temperature,
            do_sample=True
        )
        
        # 在后台线程运行生成
        thread = Thread(target=self.model.generate, kwargs=generation_kwargs)
        thread.start()
        
        for text in streamer:
            yield text
        
        thread.join()

# API端点
class GenerateRequest(BaseModel):
    prompt: str
    max_tokens: int = 512
    temperature: float = 0.7
    stream: bool = False

@app.post("/generate")
async def generate(request: GenerateRequest):
    service = LLMService()
    
    if request.stream:
        return StreamingResponse(
            service.generate(**request.dict()),
            media_type="text/event-stream"
        )
    else:
        result = ""
        async for chunk in service.generate(**request.dict()):
            result += chunk
        return {"response": result}

架构决策总结

总结

AI应用架构的成功实施，需要在技术选型、成本控制、安全合规之间找到平衡。RAG模式解决了知识注入和幻觉问题，Agent模式实现了复杂任务的自主执行，微调模式则提供了领域适应的精细控制。

架构师的核心职责是理解业务需求，选择合适的技术组合，并建立可靠的生产运维体系。AI不是银弹，但在正确的架构下，它能成为业务创新的强大引擎。