大模型应用的架构演进

大模型应用正在从简单的API封装走向复杂的Agent系统。理解架构模式的演进,有助于为不同复杂度的需求选择合适的设计。

架构模式全景

模式一:直连API(Level 0)

最简单的应用——直接调用LLM API:

用户输入 → LLM API → 输出
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": user_input}]
)

适用: 简单问答、单轮交互 优点: 开发成本最低,几天上线 缺点: 无上下文管理、无工具、无定制

模式二:对话管理(Level 1)

加入对话历史管理:

用户输入 → [+对话历史] → LLM → 输出
         历史管理(摘要/截断)
class ChatManager:
    def __init__(self, max_history=10):
        self.history = []
        self.max_history = max_history
    
    async def chat(self, user_input):
        messages = self.history + [{"role": "user", "content": user_input}]
        response = await llm.generate(messages)
        self.history.append({"role": "user", "content": user_input})
        self.history.append({"role": "assistant", "content": response})
        
        if len(self.history) > self.max_history * 2:
            self.history = self.summarize_old_history()
        
        return response

适用: 聊天机器人、通用助手 新增能力: 多轮对话、上下文记忆

模式三:RAG增强(Level 2)

加入检索增强:

用户输入 → 检索知识库 → [用户输入 + 检索结果] → LLM → 输出
class RAGApplication:
    def __init__(self, vector_db, llm):
        self.db = vector_db
        self.llm = llm
    
    async def answer(self, question):
        # 检索
        docs = self.db.search(question, top_k=5)
        context = "\n".join(doc.content for doc in docs)
        
        # 生成
        prompt = f"""基于以下信息回答问题:
        参考资料:
        {context}
        
        问题: {question}
        """
        return await self.llm.generate(prompt)

适用: 知识问答、文档问答、企业知识库 新增能力: 外部知识接入、事实性增强

模式四:工具增强(Level 3)

加入工具调用:

用户输入 → LLM → [需要工具?] 
               ├─ 是 → 调用工具 → [结果] → LLM → 输出
               └─ 否 → 直接输出
class ToolAugmentedApp:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
    
    async def run(self, user_input):
        messages = [{"role": "user", "content": user_input}]
        
        while True:
            response = await self.llm.generate(messages, tools=self.tools)
            
            if response.tool_calls:
                for call in response.tool_calls:
                    result = await self.execute_tool(call)
                    messages.append({"role": "tool", "content": result})
            else:
                return response.content

适用: 需要实时数据、外部操作的场景 新增能力: 信息获取、操作执行

模式五:Agent系统(Level 4)

自主规划和执行:

用户输入 → [规划器] → [执行器] → [验证器] → 输出
              ↑                          |
              └──── 反馈循环 ────────────┘
class AgentSystem:
    def __init__(self, planner, executor, validator, tools):
        self.planner = planner
        self.executor = executor
        self.validator = validator
        self.tools = tools
    
    async def run(self, task):
        # 1. 规划
        plan = await self.planner.create_plan(task)
        
        # 2. 执行
        for step in plan.steps:
            result = await self.executor.execute(step, self.tools)
            
            # 3. 验证
            valid = await self.validator.check(step, result)
            
            if not valid:
                # 4. 重规划
                plan = await self.planner.revise_plan(task, step, result)
        
        return await self.synthesize_results(plan)

适用: 复杂任务、多步推理、自主决策 新增能力: 任务规划、自主执行、错误恢复

模式六:多Agent协作(Level 5)

多个专精Agent协作:

[协调者] ←→ [研究Agent]
         ←→ [分析Agent]  
         ←→ [写作Agent]
         ←→ [审核Agent]
class MultiAgentSystem:
    def __init__(self):
        self.coordinator = Agent("coordinator")
        self.specialists = {
            "researcher": Agent("researcher"),
            "analyst": Agent("analyst"),
            "writer": Agent("writer"),
            "reviewer": Agent("reviewer")
        }
    
    async def run(self, task):
        # 协调者分配任务
        subtasks = await self.coordinator.decompose(task)
        
        results = {}
        for subtask in subtasks:
            agent = self.specialists[subtask.assigned_role]
            results[subtask.id] = await agent.run(subtask)
        
        # 协调者汇总
        final_result = await self.coordinator.synthesize(results)
        
        # 审核者检查
        review = await self.specialists["reviewer"].run(final_result)
        if review.needs_revision:
            return await self.run(task)  # 重做
        
        return final_result

适用: 复杂项目、需要多视角的任务 新增能力: 角色分工、协作、互相审查

架构选择决策树

需求复杂度?
├─ 单轮问答 → Level 0 (直连API)
├─ 多轮对话 → Level 1 (对话管理)
├─ 需要领域知识 → Level 2 (RAG增强)
├─ 需要外部数据/操作 → Level 3 (工具增强)
├─ 需要自主完成复杂任务 → Level 4 (Agent系统)
└─ 需要多角色协作 → Level 5 (多Agent)

生产级架构

系统架构图

                    [用户端]
                  [API网关]
              [认证&限流&路由]
              [Agent编排引擎]
               ↓         ↓
          [RAG引擎]  [工具服务]
          ├──向量库  ├──API连接器
          ├──知识库  ├──代码沙箱
          └──重排器  └──文件系统
          [模型推理层]
          ├──大模型API/自部署
          └──模型路由
          [监控&审计]
          ├──调用追踪
          ├──成本统计
          └──质量评估

关键设计原则

1. 可扩展性

  • 每个组件可独立扩展
  • 无状态服务(状态外部化)
  • 异步处理(消息队列解耦)

2. 可靠性

  • 模型服务降级方案
  • 工具调用重试机制
  • 熔断器防止级联故障

3. 可观测性

  • 全链路追踪
  • 实时监控仪表盘
  • 质量评估Pipeline

4. 安全性

  • 用户认证授权
  • 敏感数据脱敏
  • 操作审计日志
  • Prompt注入防护

常见架构反模式

反模式1:万能Agent

问题: 把所有功能塞进一个Agent
后果: Prompt巨长、上下文爆炸、难以维护

解决: 按职责拆分为多个Agent

反模式2:同步阻塞

问题: 等LLM完整生成再返回
后果: 用户等太久

解决: 流式返回,首token尽快输出

反模式3:硬编码流程

问题: 把业务逻辑硬编码在Agent中
后果: 每次需求变化都要改代码

解决: 配置化流程,规则引擎外置

总结

大模型应用架构没有"最佳实践"——只有"最适合的实践"。从Level 0的简单API调用到Level 5的多Agent协作,每个层级解决不同复杂度的问题。核心原则是:用最简单的架构解决需求——不要为了用Agent而用Agent。Level 2的RAG系统可能解决80%的企业需求,不需要复杂的Agent架构。当简单架构不够用时,再逐步升级。