Agent架构五层模型：从LLM内核到交互界面

为什么你的Agent代码越写越乱？因为你缺少一张清晰的架构地图。

---

开篇钩子

很多开发者入门Agent时，会直接把LLM API调用、工具函数、记忆存储、用户界面全塞在一个文件里。几百行代码后，逻辑纠缠成一团乱麻，加一个功能要改五处，调试时更是无从下手。Agent开发不是简单的"调用API+写Prompt"，它需要系统性的架构设计。本文提出的五层模型，将Agent系统从底层到顶层拆解为五个清晰的层次，帮你建立可扩展、可维护的Agent工程思维。

---

核心概念

Agent系统是一个复杂的软件工程问题，涉及模型推理、工具集成、状态管理、任务调度、用户交互等多个维度。五层模型将这些问题域垂直切分，每一层只关注一个核心职责，层与层之间通过定义良好的接口通信。

五层模型总览

层级	名称	核心职责	典型技术
L1	模型层（Model Layer）	LLM推理与响应生成	GPT-4, Claude, Qwen, DeepSeek
L2	能力层（Capability Layer）	工具定义、注册与执行	Function Calling, MCP, API, skills
L3	记忆层（Memory Layer）	短期与长期记忆管理	上下文窗口, 向量数据库
L4	编排层（Orchestration Layer）	任务规划、调度与执行控制	ReAct, Plan-and-Execute
L5	交互层（Interaction Layer）	用户界面与输入输出处理	Web, CLI, Bot

这种分层设计的核心价值在于：关注点分离。你可以独立替换L1的模型（从GPT-4切到Claude），而不影响其他层；可以独立扩展L2的工具集，无需改动编排逻辑；可以为同一个Agent核心（L1-L4）提供Web界面（L5），也可以提供CLI界面。

各层之间的关系

L5（交互层）接收用户输入，传递给L4（编排层）；L4制定执行计划，在需要时调用L1（模型层）进行推理，调用L2（能力层）执行工具，调用L3（记忆层）存取上下文；执行结果逐层返回，最终通过L5呈现给用户。

---

原理图解

数据流向：用户请求从L5进入，经L4分解为子任务；L4调用L1进行推理，调用L2执行工具；L3在全程提供记忆支持；执行结果沿原路返回，经L5输出给用户。

---

代码实战

下面用Python实现一个基于五层模型的Agent框架骨架，展示各层如何独立封装、协同工作。

from abc import ABC, abstractmethod
from typing import Dict, List, Any, Optional
import json

# ============================================
# L1: 模型层 - 负责LLM推理
# ============================================

class BaseModel(ABC):
    """模型层抽象基类"""
    
    @abstractmethod
    def generate(self, prompt: str, tools: List[Dict] = None) -> Dict[str, Any]:
        """生成推理结果，返回包含thought/action的结构"""
        pass


class MockModel(BaseModel):
    """模拟模型实现（真实场景替换为OpenAI/Claude API）"""
    
    def generate(self, prompt: str, tools: List[Dict] = None) -> Dict[str, Any]:
        # 模拟LLM推理：根据prompt内容返回不同的ReAct响应
        if "天气" in prompt:
            return {
                "thought": "需要查询天气",
                "action": "get_weather",
                "action_input": {"city": "北京"}
            }
        elif "计算" in prompt:
            return {
                "thought": "需要执行计算",
                "action": "calculate",
                "action_input": {"expr": "100 + 200"}
            }
        return {
            "thought": "任务完成",
            "action": "finish",
            "action_input": {}
        }


# ============================================
# L2: 能力层 - 负责工具管理
# ============================================

class ToolRegistry:
    """工具注册中心：统一管理所有可用工具"""
    
    def __init__(self):
        self._tools: Dict[str, Dict] = {}  # 工具元数据
        self._handlers: Dict[str, callable] = {}  # 工具执行函数
    
    def register(self, name: str, description: str, 
                 parameters: Dict, handler: callable):
        """注册新工具"""
        self._tools[name] = {
            "name": name,
            "description": description,
            "parameters": parameters
        }
        self._handlers[name] = handler
    
    def get_tool_schemas(self) -> List[Dict]:
        """获取所有工具的Schema（供LLM使用）"""
        return list(self._tools.values())
    
    def execute(self, name: str, params: Dict) -> Any:
        """执行指定工具"""
        if name not in self._handlers:
            raise ValueError(f"工具 '{name}' 未注册")
        return self._handlers[name](**params)


# 定义具体工具函数
def get_weather(city: str) -> str:
    """查询天气工具"""
    return f"{city}今天晴天，25°C"


def calculate(expr: str) -> str:
    """计算工具"""
    try:
        result = eval(expr, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"


# ============================================
# L3: 记忆层 - 负责上下文管理
# ============================================

class MemoryManager:
    """记忆管理器：短期记忆 + 长期记忆接口"""
    
    def __init__(self, max_short_term: int = 10):
        self.short_term: List[Dict] = []  # 短期记忆：最近N轮交互
        self.max_short_term = max_short_term
    
    def add(self, role: str, content: str, metadata: Dict = None):
        """添加记忆条目"""
        entry = {"role": role, "content": content}
        if metadata:
            entry["metadata"] = metadata
        self.short_term.append(entry)
        # 保持短期记忆容量上限
        if len(self.short_term) > self.max_short_term:
            self.short_term.pop(0)
    
    def get_context(self) -> str:
        """获取格式化的上下文字符串"""
        lines = []
        for entry in self.short_term:
            lines.append(f"[{entry['role']}] {entry['content']}")
        return "\n".join(lines)
    
    def clear(self):
        """清空短期记忆"""
        self.short_term = []


# ============================================
# L4: 编排层 - 负责任务调度
# ============================================

class AgentOrchestrator:
    """Agent编排器：核心调度逻辑"""
    
    def __init__(self, model: BaseModel, tools: ToolRegistry, 
                 memory: MemoryManager, max_steps: int = 5):
        self.model = model
        self.tools = tools
        self.memory = memory
        self.max_steps = max_steps
    
    def run(self, user_input: str) -> str:
        """执行完整任务流程"""
        # 将用户输入存入记忆
        self.memory.add("user", user_input)
        
        for step in range(self.max_steps):
            print(f"\n[Step {step + 1}/{self.max_steps}]")
            
            # 构建包含上下文的Prompt
            context = self.memory.get_context()
            tool_schemas = self.tools.get_tool_schemas()
            prompt = self._build_prompt(context, tool_schemas)
            
            # L1: 调用模型推理
            response = self.model.generate(prompt, tool_schemas)
            thought = response.get("thought", "")
            action = response.get("action", "")
            action_input = response.get("action_input", {})
            
            print(f"  Thought: {thought}")
            print(f"  Action: {action}({action_input})")
            
            # 将思考存入记忆
            self.memory.add("assistant", thought, 
                          {"action": action, "input": action_input})
            
            # 检查是否结束
            if action == "finish":
                return "任务完成"
            
            # L2: 执行工具
            try:
                observation = self.tools.execute(action, action_input)
            except Exception as e:
                observation = f"执行错误: {e}"
            
            print(f"  Observation: {observation}")
            
            # 将观察结果存入记忆
            self.memory.add("system", observation, 
                          {"step": step, "action": action})
        
        return "达到最大步数限制"
    
    def _build_prompt(self, context: str, tools: List[Dict]) -> str:
        """构建发送给LLM的完整Prompt"""
        tool_desc = json.dumps(tools, ensure_ascii=False, indent=2)
        return f"""你是一个智能助手，可以使用以下工具：
{tool_desc}

历史记录：
{context}

请按照以下格式回复：
Thought: 你的思考过程
Action: 你要调用的工具名
Action Input: 工具参数（JSON格式）

如果需要结束任务，Action请填写"finish"。"""


# ============================================
# L5: 交互层 - 负责用户界面
# ============================================

class CLIInterface:
    """命令行交互界面"""
    
    def __init__(self, orchestrator: AgentOrchestrator):
        self.orchestrator = orchestrator
    
    def start(self):
        """启动交互循环"""
        print("=" * 50)
        print("Agent CLI 已启动，输入 'quit' 退出")
        print("=" * 50)
        
        while True:
            user_input = input("\n你: ").strip()
            if user_input.lower() in ("quit", "exit", "q"):
                print("再见！")
                break
            
            result = self.orchestrator.run(user_input)
            print(f"\nAgent: {result}")


# ============================================
# 组装与运行
# ============================================

def create_agent() -> CLIInterface:
    """工厂函数：组装五层Agent系统"""
    
    # L1: 初始化模型
    model = MockModel()
    
    # L2: 注册工具
    tools = ToolRegistry()
    tools.register(
        name="get_weather",
        description="查询指定城市的天气",
        parameters={"city": "城市名称"},
        handler=get_weather
    )
    tools.register(
        name="calculate",
        description="计算数学表达式",
        parameters={"expr": "数学表达式"},
        handler=calculate
    )
    
    # L3: 初始化记忆
    memory = MemoryManager(max_short_term=10)
    
    # L4: 组装编排器
    orchestrator = AgentOrchestrator(
        model=model,
        tools=tools,
        memory=memory,
        max_steps=5
    )
    
    # L5: 创建交互界面
    interface = CLIInterface(orchestrator)
    
    return interface


if __name__ == "__main__":
    agent = create_agent()
    
    # 运行单条测试
    print("\n[测试] 查询天气")
    agent.orchestrator.run("北京天气怎么样？")
    
    print("\n" + "=" * 50)
    print("[测试] 数学计算")
    agent.orchestrator.memory.clear()  # 清空记忆
    agent.orchestrator.run("帮我计算100加200")

代码解读：

1. L1模型层：BaseModel抽象基类定义了统一接口，MockModel是模拟实现。真实项目中，你可以创建OpenAIModel、ClaudeModel等具体类，无缝切换底层模型。

2. L2能力层：ToolRegistry实现了工具的注册、发现和执行。新增工具只需调用register()，无需改动其他代码。工具Schema自动转换为LLM可理解的格式。

3. L3记忆层：MemoryManager管理短期记忆（最近N轮对话）。可以扩展为支持长期记忆（接入向量数据库如ChromaDB、Milvus）。

4. L4编排层：AgentOrchestrator是核心控制器，协调模型、工具和记忆。它实现了ReAct循环，并负责Prompt构建和步数控制。

5. L5交互层：CLIInterface提供命令行界面。可以平行创建WebInterface（基于FastAPI/Gradio）或BotInterface（基于钉钉/飞书机器人），共享同一套L1-L4核心。

---

踩坑指南

坑1：层间耦合过紧

现象：模型层直接调用工具执行函数，绕过能力层；或者编排层直接操作数据库，绕过记忆层。

解决：严格遵守"每层只依赖下层接口"的原则。使用依赖注入（如create_agent工厂函数）而非全局变量，确保各层通过抽象接口交互。

坑2：Prompt散落在各处

现象：每个模块都拼接自己的Prompt字符串，导致维护困难、风格不一致。

解决：在L1模型层集中管理Prompt模板，使用Jinja2等模板引擎。所有Prompt变更只需修改一处。

坑3：工具执行阻塞主线程

现象：某个工具调用耗时很长（如爬取网页），导致整个Agent卡住。

解决：L2能力层使用异步执行（asyncio）或线程池。编排层设置工具调用超时，超时时返回友好提示而非崩溃。

坑4：记忆膨胀拖慢推理

现象：长期对话后，记忆内容过长，每次LLM调用消耗的Token激增，响应变慢。

解决：L3记忆层实现智能摘要机制——当记忆超过阈值时，用LLM对历史对话做压缩摘要，只保留关键信息。

---

进阶思考

1. 微服务化Agent架构

当Agent系统需要服务大量用户时，五层模型可以映射为微服务架构：L1模型层部署为独立的推理服务（支持GPU集群），L3记忆层使用分布式缓存（Redis）和向量数据库服务，L4编排层作为无状态服务水平扩展。这种架构下，L5交互层可以是Web、App、IoT设备等多种形态。

2. 编排层的进化：从ReAct到DAG

简单任务用ReAct循环即可，但复杂业务流程（如"审批-执行-通知"）需要更强大的编排能力。未来趋势是将工作流定义为有向无环图（DAG），L4编排层变成图执行引擎，支持条件分支、并行执行、人工审批节点等。

3. 记忆层的分层存储策略

短期记忆、工作记忆、长期记忆、情景记忆——人脑有多级记忆系统，Agent也可以借鉴。短期记忆用上下文窗口（最近3轮），工作记忆用结构化存储（当前任务状态），长期记忆用向量数据库（知识库），情景记忆用图数据库（实体关系）。

4. 能力层的标准化：MCP协议

Anthropic推出的Model Context Protocol（MCP）正在推动工具调用的标准化。未来L2能力层可能不再维护私有的工具注册表，而是接入标准化的MCP服务器生态，实现"一次接入，处处可用"。

---

文末互动

你的Agent项目目前是"大杂烩"风格，还是已经做了分层设计？遇到过哪些架构层面的痛点？欢迎在评论区交流。下一篇，我们将聚焦L1模型层，深入探讨LLM作为Agent"大脑"的能力边界与选型策略，帮你选对模型、用好模型。

---

本文是「AI Agent编程学习系列」第2篇，关注公众号获取更多实战干货。