Agent架构设计：核心模块与数据流解析

1. 总览：Agent 的核心模块与数据流

Agent 架构的本质是构建一个能理解、规划和执行复杂任务的智能系统。在 MOSAIC 项目中，这个系统被拆解为三个核心轴线：

1. 显式命令路由 - 处理类似 /diary 这样的结构化指令
2. 自然语言对话 - 解析自由格式的用户输入
3. 后台记忆处理 - 异步构建知识索引

这种分离的设计带来了几个关键优势：

• 命令路由可以快速响应明确需求
• 对话系统能处理模糊意图
• 后台处理不阻塞用户交互

提示：虽然示例用日记场景演示，但同样的架构适用于客服、编程助手等任何需要复杂交互的场景。

2. 路由系统：意图识别与任务分发

路由层相当于 Agent 的"前台接待"，核心职责是将用户输入映射到具体的处理模块。在 MOSAIC 中，路由实现包含三个关键设计：

2.1 命令字映射表

python复制command_map = {
    '/diary': DiaryAgent,
    '/ideas': IdeasAgent,
    '/mood': MoodAnalyzer
}

这种显式映射保证了高频功能的快速响应，避免了不必要的LLM调用开销。

2.2 自然语言参数解析

即使用户使用命令字，后面的参数仍可能是自由文本。例如：

code复制/diary 今天和团队讨论了项目架构，感觉找到了新方向

路由层需要提取出结构化信息（这里是"diary"类型）并将剩余文本传递给对应处理器。

2.3 后备机制

当输入不匹配任何命令时，系统会fallback到对话Agent。这种分层处理既保证了效率，又不损失灵活性。

3. 记忆系统的三层架构

Agent的记忆不是简单的聊天记录保存，而是分层的知识管理系统：

3.1 短期记忆实现

通过ConversationState类维护对话上下文：

python复制class ConversationState:
    def __init__(self):
        self.history = []  # 保存对话轮次
        self.temp_vars = {}  # 临时变量存储

3.2 长期记忆优化

使用SQLite的JSON字段存储结构化数据，既保证查询效率，又保持灵活性：

sql复制CREATE TABLE diary_entries (
    id INTEGER PRIMARY KEY,
    created_at TIMESTAMP,
    content TEXT,
    metadata JSON  -- 存储情绪标签等扩展数据
);

3.3 向量记忆构建

后台Processor定期执行：

1. 扫描新增文本
2. 调用LLM生成摘要和关键词
3. 存入向量数据库

python复制def process_new_entries():
    new_items = db.get_unprocessed()
    for item in new_items:
        embedding = llm.embed(item.text)
        vector_db.upsert(item.id, embedding, metadata)
        db.mark_processed(item.id)

4. RAG的实战应用模式

在MOSAIC中，检索增强生成(RAG)不是简单的"问答-检索"模式，而是深度集成到工作流中：

4.1 主动检索场景

当用户提及历史内容时触发：

code复制"上次提到的项目架构想法是什么？"
→ 自动检索最近7天包含"架构"的ideas
→ 将相关片段注入prompt

4.2 被动增强场景

即使用户没有明确要求，系统也会在以下情况自动检索：

• 检测到模糊指代（"那个方法"）
• 时间相关表述（"上周"）
• 情绪性表达（"之前困扰我的问题"）

4.3 混合检索策略

结合三种检索方式：

1. 关键词搜索（SQL LIKE）
2. 时间过滤（WHERE created_at > ?）
3. 向量相似度（ORDER BY embedding_distance）

5. 工具调用机制详解

Agent的工具系统设计遵循三个原则：

5.1 声明与实现分离

每个工具需要提供：

• 描述（LLM可见）
• 参数schema
• 执行函数

python复制@tool
def search_contacts(name: str) -> List[Contact]:
    """根据姓名查找联系人"""
    return db.query(Contact).filter_by(name=name).all()

5.2 动态权限控制

通过装饰器实现工具可见性管理：

python复制@require_role('admin')
def delete_entry(entry_id: int):
    """删除日记条目"""
    db.delete(Entry).where(id=entry_id)

5.3 执行监控

记录工具调用的完整审计日志：

python复制def tool_wrapper(func):
    def inner(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            log_tool_usage(func.__name__, 'success', time.time()-start)
            return result
        except Exception as e:
            log_tool_usage(func.__name__, str(e), time.time()-start)
            raise
    return inner

6. 技能系统的设计哲学

MOSAIC采用"文档+脚本"的技能开发模式，而非纯LLM生成代码，这是经过实践验证的更可靠方案：

6.1 技能模板结构

code复制skills/
    summarize_diary/
        SKILL.md       # 自然语言描述
        requirements.txt # 依赖库
        main.py        # 可测试的独立脚本

6.2 与普通工具的区别

1. 技能可以组合多个工具
2. 包含更复杂的控制流
3. 有独立的错误处理机制

6.3 开发流程

1. 人工编写技能文档
2. 实现基础脚本
3. 注册到技能库
4. Agent通过描述学习调用方式

7. 任务规划的三种模式

规划器是Agent的"大脑皮层"，负责分解复杂请求：

7.1 线性规划

适用于明确步骤的任务：

code复制"记录今天的跑步数据并分析趋势"
→ 1. 调用log_exercise记录数据
→ 2. 调用analyze_trends生成报告

7.2 条件规划

根据中间结果决定后续步骤：

python复制if weather := check_weather(location):
    suggest_activity(weather)
else:
    ask_clarification()

7.3 动态调整

在执行过程中修正计划：

python复制while not task.is_complete():
    next_step = planner.replan(current_state)
    execute(next_step)
    current_state = get_updated_state()

8. 多轮对话状态管理

对话系统的核心挑战是维护上下文一致性：

8.1 对话状态机

mermaid复制stateDiagram
    [*] --> Idle
    Idle --> CollectingInfo : 需要补充参数
    CollectingInfo --> Executing : 参数齐全
    Executing --> Confirming : 需要用户确认
    Confirming --> Executing : 用户确认
    Confirming --> CollectingInfo : 用户修改

8.2 上下文修补机制

当用户突然切换话题时：

1. 检测话题偏移（通过嵌入相似度）
2. 显式确认是否要切换
3. 必要时创建新的对话分支

8.3 超时处理

对话状态保存策略：

• 活跃会话：内存驻留
• 闲置15分钟：持久化到DB
• 超过1天：归档压缩

9. 结构化输出的实现技巧

analyze模式是Agent的"理性输出通道"，用于生成机器可处理的数据：

9.1 输出约束

使用Pydantic模型定义输出结构：

python复制class SentimentAnalysis(BaseModel):
    overall: Literal['positive', 'neutral', 'negative']
    aspects: Dict[str, float]  # 各维度评分
    summary: str

9.2 提示词工程

强制JSON输出模式：

python复制prompt = """严格按照以下schema输出JSON：
{schema}

输入文本：{text}"""

9.3 后处理验证

python复制def validate_output(raw: str, model: Type[BaseModel]) -> BaseModel:
    try:
        data = json.loads(raw)
        return model.parse_obj(data)
    except Exception as e:
        log_validation_error(e)
        raise OutputValidationError()

10. 架构演进建议

基于MOSAIC的实践，当你要设计自己的Agent系统时：

1. 从数据流开始设计 - 先明确信息如何在各模块间流动
2. 隔离LLM依赖 - 将模型调用封装为独立服务
3. 重视可观测性 - 记录完整的决策链路
4. 设计降级方案 - 当LLM不可用时能回退到规则系统

示例项目中的llm_proxy.py展示了如何构建模型调用中间层：

python复制class LLMProxy:
    def __init__(self):
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )

    @circuit_breaker
    def chat_completion(self, messages):
        # 统一处理重试、限流、fallback等

附录：MOSAIC实现细节

后端架构要点

• 单进程FastAPI服务
• 同步处理HTTP请求
• 异步执行后台任务
• 结构化日志记录

前端交互设计

• 静态页面+Alpine.js
• 流式响应处理
• 本地存储缓存
• 移动端适配

部署注意事项

1. 向量数据库需要定期维护（重建索引）
2. LLM调用需要实现速率限制
3. 对话状态存储要考虑内存限制
4. 开发环境与生产环境的模型配置差异

性能优化技巧

• 对话历史摘要（而非完整保存）
• 向量检索的预过滤
• 工具调用的并行化
• 模型输出的确定性控制

我在实际开发中发现，Agent系统的复杂度主要来自状态管理而非模型调用。建议在初期就建立完善的：

• 请求追踪系统（trace_id贯穿所有日志）
• 性能监控看板
• 异常分类处理机制

一个实用的调试技巧是保存完整的"思考链"快照：

python复制def debug_agent_thoughts():
    snapshot = {
        'input': latest_input,
        'internal_state': agent.get_state(),
        'llm_calls': llm_proxy.get_logs(),
        'tool_calls': tool_manager.get_logs()
    }
    save_debug_snapshot(snapshot)