browse.nuvatech.cn Wiki 导出

• - 导出时间: 2026-05-08T02:40:37.206089759Z
• - Wiki ID: 2026-05-08-104037
• - 来源: https://browse.nuvatech.cn

目录

• - 项目概览
• - 环境搭建与启动
• - 环境变量配置
• - 配置文件结构
• - 系统架构总览
• - 单主 Agent 架构
• - 技能加载与管理
• - 技能工具门控机制
• - Dialog State 设计
• - SOP 阶段推进
• - 用户确认机制
• - 工具注册与工厂
• - 工具描述管理
• - 工具调用链路追踪
• - 消息入站处理流程
• - 微信客服与小程序集成
• - 会话分发与回复状态
• - Mem0 记忆管理
• - 用户洞察分析
• - 穿搭工作流服务
• - 图片生成与处理
• - 单品初始化处理
• - Dashboard 管理面板
• - Admin API 接口
• - 链路追踪与 Trace
• - 系统监控与健康检查
• - 单元测试与集成测试
• - 自动化测试与修复
• - 部署与发布
• - 新增业务工具
• - 新增技能模块
• - 提示词管理与国际化

<a id="1-xiang-mu-gai-lan"></a>

---

项目概览

• - Section: 快速入门
• - Source File: 1-xiang-mu-gai-lan.md

outfit_agent 是一个基于 Agno 框架构建的智能穿搭顾问 REST API 服务。系统采用单主 Agent + 按需加载 Skill + SOP 状态推进的架构模式，为微信客服和微信小程序渠道提供专业的穿搭建议、单品管理和造型生成服务。

系统定位

outfit_agent 的核心定位是穿搭领域的智能顾问，而非通用聊天机器人。系统通过技能（Skill）机制实现专业能力的模块化加载，在穿搭构想、用户确认、搜索单品、穿搭输出等关键节点与用户进行显式交互，确保每一步决策都经过用户确认。

项目依赖同目录下的 outfit_biz 包，该包封装了穿搭业务的核心逻辑，包括数据库模型、业务工具函数和配置管理。outfit_agent 在其基础上构建 Agent 运行时、消息处理和渠道集成。

Sources: README.md, AGENTS.md, docs/agent_refact.md

技术架构

graph TB

    subgraph "外部渠道"

        WK[微信客服]

        WM[微信小程序]

    end

    

    subgraph "outfit_agent 服务"

        direction TB

        FA[FastAPI 应用]

        AO[AgentOS]

        MA[主 Agent]

        

        subgraph "技能系统"

            SK1[style-advisor-skill]

            SK2[outfit-output-skill]

            SK3[buyer-wardrobe-skill]

            SK4[nuva-soul]

        end

        

        subgraph "工具系统"

            T1[穿搭工具]

            T2[用户工具]

            T3[图片工具]

            T4[消息工具]

        end

        

        subgraph "服务层"

            WS[工作流服务]

            MS[消息服务]

            IS[洞察服务]

        end

    end

    

    subgraph "数据层"

        PG[(PostgreSQL)]

        RD[(Redis)]

        M0[(Mem0)]

    end

    

    subgraph "外部服务"

        LLM[LLM 模型]

        OSS[对象存储]

        WX[微信 API]

    end

    

    WK --> FA

    WM --> FA

    FA --> AO

    AO --> MA

    MA --> SK1 & SK2 & SK3 & SK4

    SK1 & SK2 & SK3 & SK4 --> T1 & T2 & T3 & T4

    MA --> WS & MS & IS

    WS & MS & IS --> PG & RD & M0

    MA --> LLM

    T3 --> OSS

    MS --> WX

系统采用分层架构设计：

入口层：FastAPI 应用通过路由模块接收来自不同渠道的请求，支持普通 JSON 响应和 SSE 流式响应。

Agent 运行时：基于 Agno 框架的 Agent 单例，负责意图识别、技能加载和工具调用。AgentOS 提供会话持久化、链路追踪和调度能力。

技能系统：通过 TOML 配置文件定义技能，每个技能包含工具列表、指令提示和激活条件。技能按需加载，与 SOP 阶段绑定。

工具系统：业务工具函数通过工厂模式注册，支持渠道门控和技能绑定。工具描述通过 descriptions.toml 管理，支持灰度覆盖。

服务层：工作流服务管理穿搭生成的长流程，消息服务处理多渠道消息分发，洞察服务负责用户画像的离线分析。

数据层：PostgreSQL 存储会话、用户和业务数据，Redis 提供缓存和队列能力，Mem0 管理长期语义记忆。

Sources: app/main.py, app/agentos.py, app/agent.py

核心模块

Agent 运行时

Agent 运行时是系统的核心，采用单例模式确保实例复用。主 Agent 负责意图识别和任务分发，通过技能机制动态加载专业能力。

# Agent 单例初始化

_agent: Agent | None = None



def init_agent() -> Agent:

    """初始化并返回 Agent 单例"""

    global _agent

    if _agent is not None:

        return _agent

    # ... 构建 Agent 实例

    return _agent

Agent 配置通过 data/team/main.toml 定义，包括可用工具、技能列表和渠道专属工具。模型配置通过 config.yml 的 model_map 节管理，支持多模型灰度。

Sources: app/agent.py, data/team/main.toml

技能系统

技能系统实现专业能力的模块化管理。每个技能通过 skill.toml 定义元数据、工具列表和激活条件。

# 技能定义示例

id = "style-advisor-skill"

name = "Style Advisor"

description = "风格顾问技能"

activation = "on_demand"



[metadata]

category = "styling"

owner = "main-team"



[instructions]

zh = "skills/style_advisor.zh.txt"

en = "skills/style_advisor.en.txt"

技能支持以下特性：

• - 按需激活：根据用户意图和 SOP 阶段动态加载
• - 渠道门控：某些工具仅在特定渠道可用
• - 灰度控制：通过灰度配置服务控制技能对用户的可见性
• - 指令注入：技能指令通过 PromptLoader 从文件加载

Sources: app/skills/loader.py, data/skills/nuva-soul/skill.toml

工具系统

工具系统采用工厂模式，支持工具注册、描述管理和运行时门控。

# 工具注册示例

def materialize_tools_for_user(

    tools: Sequence[Function | Callable[..., Any]],

    *,

    user_id: int | str | None = None,

) -> list[Function | Callable[..., Any]]:

    """为用户物化工具列表，应用描述和门控规则"""

    # ...

工具通过 app/tools/ 目录下的模块定义，每个工具函数必须包含完整的 docstring，Agno 会据此生成函数 schema 供 LLM 调用。

工具描述通过 data/tools/descriptions.toml 集中管理，支持按用户灰度覆盖。

Sources: app/tools/factory.py, data/tools/descriptions.toml

消息处理

消息处理模块负责多渠道消息的接收、分发和回复。支持文本、图片和富媒体消息。

# 消息分发示例

async def dispatch_text(

    text: str,

    *,

    tool_ctx: ToolContext,

    channel: str,

    # ...

) -> dict[str, Any]:

    """分发文本消息到 Agent 处理"""

    # ...

消息处理流程：

1. 1. 入站处理：验证、审计、快捷回复检测
2. 2. 会话分发：根据会话状态决定处理方式
3. 3. Agent 调用：将消息发送给 Agent 处理
4. 4. 回复格式化：根据渠道特性格式化回复
5. 5. 消息发送：通过渠道 API 发送回复

Sources: app/messaging/__init__.py, app/messaging/session_dispatcher.py

记忆与洞察

系统集成 Mem0 进行长期语义记忆管理，支持实时分析和离线洞察两种模式。

# 记忆操作示例

async def search_user_memory(

    query: str,

    user_id: str,

    limit: int = 10,

) -> list[dict[str, Any]]:

    """搜索用户相关记忆"""

    # ...

洞察服务通过定时任务分析用户行为，生成用户画像和风格偏好，用于个性化推荐。

Sources: app/memory/__init__.py, app/insights/__init__.py

数据流

系统处理用户请求的典型流程如下：

sequenceDiagram

    participant U as 用户

    participant C as 渠道

    participant F as FastAPI

    participant A as Agent

    participant S as 技能

    participant T as 工具

    participant D as 数据层

    

    U->>C: 发送消息

    C->>F: HTTP 请求

    F->>F: 入站处理（审计、快捷回复）

    F->>A: 分发到 Agent

    A->>A: 意图识别

    A->>S: 加载相关技能

    S->>T: 调用工具

    T->>D: 读写数据

    D-->>T: 返回结果

    T-->>S: 工具结果

    S-->>A: 技能结果

    A->>A: 生成回复

    A-->>F: Agent 响应

    F->>C: 格式化回复

    C->>U: 发送回复

对于穿搭生成等长流程，系统采用异步工作流模式：

1. 1. 任务创建：创建穿搭生成任务，返回任务 ID
2. 2. 后台执行：工作流服务异步执行多步骤流程
3. 3. 状态更新：通过 WebSocket 或轮询向客户端推送进度
4. 4. 结果交付：完成后通过渠道 API 发送结果

Sources: app/services/outfit_workflow_service.py, app/workflows/outfit_generation.py

配置管理

系统采用分层配置管理：

环境变量：通过 .env 文件管理敏感配置，如 API 密钥、数据库连接等。

配置文件：config.yml 管理业务配置，支持 outfit_biz/config.yml 基础配置和 outfit_agent/config.yml 覆盖配置的 deep-merge。

运行时配置：通过 AgentSettings 类提供类型安全的配置访问，支持从 YAML 和环境变量读取。

# 配置读取示例

settings = get_settings()

model_cfg = settings.get_model("default")

db_url = settings.database_url

配置支持灰度控制，通过 outfit_biz.gray 模块实现按用户的配置覆盖。

Sources: app/config.py, config.yml, .env.example

部署架构

系统采用 PM2 + uvicorn 的部署方式：

// ecosystem.config.js

module.exports = {

  apps: [{

    name: 'outfit-agent',

    script: 'uvicorn',

    args: 'app.main:app --host 0.0.0.0 --port 8000',

    // ...

  }]

}

部署依赖：

• - PostgreSQL：存储会话、用户、业务数据和 AgentOS 追踪数据
• - Redis：提供缓存、队列和分布式锁能力
• - 对象存储：存储用户上传图片和生成图片
• - LLM 服务：提供大语言模型能力，支持 OpenAI 兼容接口

Sources: ecosystem.config.js, app/main.py

测试体系

项目包含完整的测试体系，覆盖单元测试、集成测试和自动化测试：

• - 单元测试：覆盖工具函数、服务逻辑和配置管理
• - 集成测试：测试 API 端点和渠道集成
• - 自动化测试：基于场景的回归测试，支持自动修复

# 运行测试

pytest



# 运行特定测试

pytest tests/test_agentos.py

测试文件位于 tests/ 目录，共 103 个测试文件，覆盖系统各个模块。

Sources: tests/, pyproject.toml

开发规范

项目遵循严格的开发规范：

代码风格：

• - 使用 Ruff 进行代码格式化和 lint 检查
• - 行长度限制为 100 字符
• - 目标 Python 版本为 3.11

提示词管理：

• - 所有提示词禁止硬编码，必须通过 PromptLoader 从文件加载
• - 用户可见文本禁止硬编码，必须通过 get_text/get_file/get_prompt 加载
• - 提示词文件位于 data/prompts/ 和 data/i18n/

Agent 实例管理：

• - Agent 实例必须单例复用，禁止在循环中反复创建
• - 工具函数必须包含完整 docstring，Agno 据此生成函数 schema

配置管理：

• - 敏感配置通过环境变量管理
• - 业务配置通过 YAML 文件管理
• - 配置变更需要重启服务生效

Sources: AGENTS.md, pyproject.toml

下一步阅读

根据您的开发需求，建议按以下顺序阅读文档：

快速入门：

• - 环境搭建与启动：了解如何本地启动项目
• - 环境变量配置：配置必要的环境变量
• - 配置文件结构：理解配置文件的组织方式

架构理解：

• - 系统架构总览：深入了解系统分层架构
• - 单主 Agent 架构：理解 Agent 运行时设计
• - 技能加载与管理：掌握技能系统的工作原理

核心功能：

• - Dialog State 设计：理解对话状态管理
• - 工具注册与工厂：了解工具系统的实现
• - 穿搭工作流服务：掌握穿搭生成流程

扩展开发：

• - 新增业务工具：学习如何添加新工具
• - 新增技能模块：学习如何创建新技能
• - 提示词管理与国际化：掌握提示词管理最佳实践

<a id="2-huan-jing-da-jian-yu-qi-dong"></a>

---

环境搭建与启动

• - Section: 快速入门
• - Source File: 2-huan-jing-da-jian-yu-qi-dong.md

本页面指导开发者从零开始搭建 outfit_agent 的本地开发/部署环境，涵盖依赖安装、配置文件设置、数据库初始化及服务启动的完整流程。

一、前置条件

在开始之前，请确保系统已安装以下基础组件：

关键数据库扩展：PostgreSQL 需安装 pgvector、pgvectorscale、pg_search 及 zhparser 扩展，用于向量检索与中文全文搜索。

Sources: pyproject.toml | README.md

二、项目结构概览

本项目采用 双仓库协作 架构：outfit_agent 依赖同级目录下的 outfit_biz 公共业务库。

graph TD

    A[outfit_agent] -->|依赖| B[outfit_biz]

    A -->|editable install| B

    C[config.yml] -->|deep-merge| D[AgentSettings]

    E[.env] -->|环境变量| D

    B -->|提供| F[数据库模型 / 服务 / 工具]

    A -->|包含| G[Agent / Skills / Routers]

    G -->|调用| F

目录结构简览：

workspace/

├── outfit_agent/          # 本项目：Agent 应用层

│   ├── app/               # 核心代码（FastAPI + Agno Agent）

│   │   ├── main.py        # 应用入口，lifespan 管理

│   │   ├── agent.py       # Agent 单例初始化

│   │   ├── config.py      # 配置加载（继承 outfit_biz.Settings）

│   │   ├── routers/       # HTTP 路由（chat/wechat/miniapp/admin）

│   │   ├── services/      # 业务服务

│   │   ├── tools/         # Agent 工具函数

│   │   ├── skills/        # 技能加载器

│   │   └── memory/        # Mem0 记忆管理

│   ├── data/              # 提示词/技能/团队配置/i18n

│   ├── dashboard/         # 前端管理面板（React + Vite）

│   ├── tests/             # 测试用例

│   ├── config.yml         # Agent 专有配置（覆盖 outfit_biz 默认值）

│   ├── .env.example       # 环境变量模板

│   ├── pyproject.toml     # Python 项目元数据

│   └── ecosystem.config.js # PM2 生产部署配置

│

└── outfit_biz/            # 共享业务库：数据模型/仓库/工具

    ├── src/outfit_biz/    # 核心包

    ├── config.yml         # 基础配置（模型映射/数据目录等）

    └── docs/init.sql      # 数据库初始化脚本

Sources: README.md | config.py

三、克隆代码

两个仓库必须放在同一父目录下（pyproject.toml 中 outfit-biz 使用了相对路径引用）：

# 进入工作目录

mkdir -p ~/outfit && cd ~/outfit



# 克隆两个仓库

git clone <outfit_agent_repo_url> outfit_agent

git clone <outfit_biz_repo_url> outfit_biz



# 验证目录结构

ls -la

# 应同时看到 outfit_agent/ 和 outfit_biz/

重要：outfit_agent/pyproject.toml 中声明了 [tool.uv.sources] outfit-biz = { path = "../outfit_biz", editable = true }，因此两个仓库的相对路径必须保持 ../outfit_biz。

Sources: pyproject.toml

四、安装 Python 依赖

本项目使用 uv 作为包管理工具，它会自动创建虚拟环境并安装所有依赖：

cd ~/outfit/outfit_agent



# 同步所有依赖（含开发依赖）

uv sync --dev

uv sync 会自动：

1. 1. 读取 pyproject.toml 和 uv.lock 中的依赖声明
2. 2. 创建 .venv 虚拟环境
3. 3. 安装 outfit_agent 自身及 outfit_biz（editable 模式）
4. 4. 安装开发工具（pytest、ruff 等）

验证安装：

# 确认 Python 版本

uv run python --version

# Python 3.12.x



# 确认关键依赖

uv run python -c "import agno; print(agno.__version__)"

uv run python -c "import fastapi; print(fastapi.__version__)"

Sources: pyproject.toml | ecosystem.config.js

五、数据库初始化

5.1 创建数据库

# 创建数据库

psql -U postgres -c "CREATE DATABASE outfit_biz;"

5.2 安装 PostgreSQL 扩展

数据库需要以下扩展来支持向量检索和中文全文搜索：

-- 连接到数据库

\c outfit_biz



-- 安装扩展

CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

CREATE EXTENSION IF NOT EXISTS vector;           -- pgvector

CREATE EXTENSION IF NOT EXISTS vectorscale;      -- pgvectorscale

CREATE EXTENSION IF NOT EXISTS pg_search;        -- BM25 全文检索

CREATE EXTENSION IF NOT EXISTS zhparser;         -- 中文分词

注意：pg_search 来自 ParadeDB 项目，需单独安装。vectorscale 是 Timescale 的高性能向量索引扩展。

5.3 初始化表结构

执行项目提供的初始化脚本：

psql -U postgres -d outfit_biz -f ../outfit_biz/docs/init.sql

该脚本会创建以下核心表：

• - users / user_profiles / user_oauth — 用户体系
• - conversations / messages — 对话与消息
• - items / images — 衣橱单品与图片（分区表）
• - outfits — 穿搭方案
• - gray_config_rules — 灰度配置
• - user_insights — 用户洞察
• - message_access_* — 消息准入控制

注意：首次运行应用时，app/main.py 的 lifespan 会自动执行增量迁移（ALTER TABLE ... ADD COLUMN IF NOT EXISTS），因此即使 init.sql 版本略有滞后，应用也能自动补齐缺失字段。

Sources: init.sql | main.py

六、配置环境变量

6.1 创建 .env 文件

cp .env.example .env

6.2 必选配置项

以下配置项为服务正常运行的最低要求：

6.3 推荐配置项

6.4 可选配置项

根据开发需求，按需配置以下模块：

企业微信接入：

WECHAT_TOKEN=your_token

WECHAT_ENCODING_AES_KEY=your_key

WECHAT_CORP_ID=your_corp_id

WECHAT_SECRET=your_secret

阿里云内容安全：

ALIYUN_AK_ID=your_access_key_id

ALIYUN_AK_SECRET=your_access_key_secret

和风天气：

QWEATHER_API_HOST=your_host.qweatherapi.com

QWEATHER_KEY_ID=your_key_id

QWEATHER_PROJECT_ID=your_project_id

QWEATHER_PRIVATE_KEY_PATH=data/ed25519-private.pem

AI 生图质量：

IMAGE_GEN_QUALITY=2K  # 可选: 1K, 2K, 4K

完整配置参考：所有可用的环境变量详见 .env.example。

Sources: .env.example | config.py

七、配置文件说明

除了环境变量，项目还使用 YAML 配置文件进行更细粒度的行为控制。配置文件采用深度合并策略：outfit_biz/config.yml 提供基础配置，outfit_agent/config.yml 在其上叠加覆盖。

7.1 config.yml 结构

graph LR

    A[outfit_biz/config.yml] -->|基础| C[合并后配置]

    B[outfit_agent/config.yml] -->|覆盖| C

    D[环境变量 .env] -->|最高优先级| C

    C --> E[AgentSettings]

优先级（从高到低）：

1. 1. 环境变量 / .env
2. 2. config.yml 合并值
3. 3. 代码中的默认值

7.2 关键配置节

7.3 模型映射示例

config.yml 中的 model_map 定义了不同角色使用的 LLM 模型：

model_map:

  default:              # 默认模型，其他角色继承其未覆盖的字段

    model: qwen3.6-plus

    provider: dashscope

    api_base: https://dashscope.aliyuncs.com/compatible-mode/v1

  flash:                # 快速模型

    model: qwen3.5-flash

  vision:               # 视觉理解模型

    model: qwen-vl-plus

    provider: dashscope

  text_embedding:       # 文本向量模型

    model: text-embedding-v4

    dimension: 1024

  embedding:            # 多模态向量模型

    model: qwen3-vl-embedding

    dimension: 512

提示：所有模型可通过 OPENAI_API_KEY + OPENAI_BASE_URL 统一走 OpenAI 兼容网关，无需为每个模型单独配置 Key。

Sources: config.yml | outfit_biz/config.yml

八、启动服务

8.1 开发模式启动

cd ~/outfit/outfit_agent



# 使用 uv 启动（自动加载 .env，开启热重载）

uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

启动成功后会看到类似日志：

Initializing Agent singleton (sync, before AgentOS construction)...

Global agno async httpx client initialized with timeout=120s

Starting up: initializing Memory (Mem0) singleton...

ImgGenHandler started

ImageCaptionHandler started

ItemProcessingHandler started

AgentOS built: tracing=True, scheduler=True, mcp=True

AgentOS get_app() complete — all routes registered.

INFO:     Uvicorn running on http://0.0.0.0:8000

8.2 生产模式启动（PM2）

项目提供了 ecosystem.config.js 用于 PM2 进程管理：

# 安装 PM2（如未安装）

npm install -g pm2



# 使用 PM2 启动

pm2 start ecosystem.config.js



# 查看日志

pm2 logs outfit-agent



# 查看状态

pm2 status

PM2 配置说明（ecosystem.config.js）：

• - 启动命令：uv run uvicorn app.main:app --host 127.0.0.1 --port 8897
• - 自动重启：autorestart: true
• - 内存限制：max_memory_restart: "1G"

8.3 启动流程图

sequenceDiagram

    participant M as main.py

    participant A as Agent Init

    participant L as Lifespan

    participant S as Services



    M->>M: load_dotenv(.env)

    M->>M: _init_httpx_client()

    M->>A: init_agent() + init_canvas_agent()

    M->>M: init_hyperdx_monitoring()

    M->>M: 注册路由 (chat/wechat/miniapp/admin)

    M->>L: AgentOS lifespan 启动

    

    Note over L: _mem0_lifespan

    L->>L: DB schema ensure (ALTER TABLE)

    L->>L: ensure_gray_config_schema()

    L->>L: ensure_insight_schema()

    L->>L: init_memory() (Mem0)

    L->>L: _init_litellm_redis_cache()

    L->>S: ImgGenHandler.start()

    L->>S: ImageCaptionHandler.start()

    L->>S: ItemProcessingHandler.start()

    L-->>M: 服务就绪

Sources: main.py | ecosystem.config.js

九、验证服务

9.1 健康检查

curl http://localhost:8000/health

# 期望返回: {"status": "ok"}

9.2 API 文档

启动后访问 Swagger UI 查看完整 API 文档：

http://localhost:8000/docs

9.3 开发模式测试请求

配置 DEV_TOKEN 和 DEV_USER_ID 后，可直接调用 /api/chat 接口：

curl -X POST http://localhost:8000/api/chat \

  -H "Content-Type: application/json" \

  -H "Authorization: Bearer your_dev_token" \

  -d '{

    "message": "你好，请帮我搭配一套春季穿搭",

    "user_id": 1

  }'

9.4 运行测试

# 运行全部测试

uv run pytest



# 运行特定测试文件

uv run pytest tests/test_agentos.py -v



# 带覆盖率

uv run pytest --cov=app --cov-report=term-missing

Sources: main.py | README.md

十、Dashboard 前端构建（可选）

Dashboard 是基于 React + Vite 的管理面板，用于调试 Agent 对话、查看 Trace 和配置管理。

cd ~/outfit/outfit_agent/dashboard



# 安装前端依赖

npm install



# 开发模式（Vite dev server）

npm run dev



# 生产构建（输出到 dist/）

npm run build

构建后，dist/ 目录会被 FastAPI 自动挂载到 /dashboard/ 路径下：

# app/main.py 中的自动挂载逻辑

_dashboard_dist = Path(__file__).resolve().parents[1] / "dashboard" / "dist"

if _dashboard_dist.exists():

    base_app.mount("/dashboard", StaticFiles(directory=str(_dashboard_dist), html=True))

开发模式下：可直接访问 http://localhost:5173（Vite 默认端口）进行前端开发，API 请求通过 Vite proxy 转发到后端。

Sources: dashboard/package.json | main.py

十一、常见问题排查

Q1: 启动时报 ModuleNotFoundError: No module named 'outfit_biz'

原因：未执行 uv sync 或 outfit_biz 目录不在 ../outfit_biz 位置。

解决：

# 确认目录结构

ls ../outfit_biz/pyproject.toml



# 重新同步依赖

uv sync --dev

Q2: 启动时报 connection to server at "localhost" ... failed

原因：PostgreSQL 未启动或连接配置错误。

解决：

# 检查 PostgreSQL 是否运行

pg_isready



# 验证连接

psql -U postgres -d outfit_biz -c "SELECT 1;"



# 确认 .env 中 DATABASE_URL 配置正确

Q3: 启动时报 Redis 连接失败

原因：Redis 未启动或地址配置错误。

解决：

# 检查 Redis 是否运行

redis-cli ping

# 期望返回: PONG



# 确认 .env 中 REDIS_URL 配置正确

Q4: LLM 调用超时

原因：默认超时 120 秒，大型模型在负载下可能需要更长时间。

解决：调整 config.yml 中的超时配置：

llm_chat_timeout_seconds: 600  # 增加到 10 分钟

Q5: pgvector 或 pg_search 扩展安装失败

原因：这些扩展需要从源码编译或使用特定发行版。

解决：

# Ubuntu/Debian: 使用 PGDG 仓库

sudo apt install postgresql-16-pgvector



# 或从源码编译 pgvector

git clone https://github.com/pgvector/pgvector.git

cd pgvector

make && sudo make install

十二、下一步

环境搭建完成后，建议按以下顺序继续阅读：

1. 1. 环境变量配置 — 详细了解每个配置项的用途和调优建议
2. 2. 配置文件结构 — 深入理解 config.yml 的合并策略和完整字段
3. 3. 系统架构总览 — 从全局视角理解系统各组件的协作关系
4. 4. 单元测试与集成测试 — 学习如何编写和运行测试
5. 5. 新增业务工具 — 扩展 Agent 能力的第一步

<a id="3-huan-jing-bian-liang-pei-zhi"></a>

---

环境变量配置

• - Section: 快速入门
• - Group: 配置与环境
• - Source File: 3-huan-jing-bian-liang-pei-zhi.md

本文档详细说明 outfit_agent 项目中所有环境变量的配置方式、用途和优先级规则。环境变量是系统配置的最高优先级来源，掌握这些变量的正确配置对于项目的正常运行至关重要。

配置加载机制

outfit_agent 的配置系统采用多层叠加架构，优先级从高到低依次为：

1. 1. 环境变量（最高优先级）：直接在操作系统或 .env 文件中设置
2. 2. .env 文件：项目根目录下的环境变量文件
3. 3. config.yml 配置文件：outfit_agent 和 outfit_biz 两个项目的 YAML 配置叠加
4. 4. 代码默认值：Python 代码中定义的字段默认值

配置加载流程如下：

flowchart TD

    A[应用启动] --> B[加载 .env 文件]

    B --> C[加载 outfit_biz/config.yml]

    C --> D[加载 outfit_agent/config.yml]

    D --> E[深度合并配置]

    E --> F[环境变量覆盖]

    F --> G[创建 AgentSettings 单例]

    

    subgraph "配置源优先级"

        H[环境变量] --> I[.env 文件]

        I --> J[config.yml]

        J --> K[代码默认值]

    end

配置加载的核心逻辑在 app/config.py 和 outfit_biz/src/outfit_biz/config.py 中实现。

Sources: app/config.py, outfit_biz/src/outfit_biz/config.py

核心环境变量清单

以下是系统运行所需的核心环境变量，按功能模块分类：

LLM 模型配置

LLM 配置推荐统一走 OpenAI 兼容网关，模型别名和默认参数在 config.yml 的 model_map 节配置。

Sources: .env.example, app/config.py

服务配置

Sources: .env.example, app/config.py

数据库与缓存配置

数据库和 Redis 是系统的核心基础设施。Mem0 组件负责长期语义记忆，完整对话历史由 agno PostgresDb 持久化。

Sources: .env.example, app/config.py, app/memory/client.py

企业微信配置

企业微信客服配置用于接收客服消息回调。小程序配置用于发送小程序卡片消息。

Sources: .env.example, app/config.py

阿里云内容安全配置

阿里云内容安全用于对用户发送的文本和图片进行合规性审核。

Sources: .env.example, app/config.py

和风天气配置

和风天气配置用于提供穿搭场景的天气查询能力。私钥文件需要从和风天气控制台生成。

Sources: .env.example, app/config.py, app/weather/client.py

图片生成配置

图片生成配置支持多个提供商，可以通过 IMAGE_GEN_PROVIDERS 设置优先级顺序。

Sources: .env.example, outfit_biz/src/outfit_biz/config.py

Admin Dashboard 配置

Admin Secret Key 是独立于用户 JWT 的管理接口认证凭证。生产环境必须设置一个随机长字符串，留空则 Admin 接口全部返回 403。

Sources: .env.example, app/config.py

监控与可观测性配置

HyperDX 监控通过 OpenTelemetry 协议发送日志数据。配置 HYPERDX_API_KEY 和 OTEL_SERVICE_NAME 后自动启用。

Sources: .env.example, app/observability.py

对象存储配置

对象存储用于存储用户上传的图片和 AI 生成的穿搭图。

Sources: outfit_biz/src/outfit_biz/config.py

穿搭工作流配置

这些配置控制穿搭工作流的异步处理行为，包括图片生成、单品初始化和图片描述等后台任务。

Sources: app/config.py, app/main.py

用户洞察配置

用户洞察是离线任务，每天定时分析用户的穿搭记录、衣橱单品和对话记录，生成用户画像并同步到 Mem0。

Sources: app/config.py, docs/insight.md

配置文件结构

环境变量与配置文件的关系如下：

graph TB

    subgraph "环境变量优先级"

        ENV[环境变量 / .env]

        YAML_BIZ[outfit_biz/config.yml]

        YAML_AGENT[outfit_agent/config.yml]

        DEFAULT[代码默认值]

    end

    

    ENV --> MERGE[深度合并]

    YAML_BIZ --> MERGE

    YAML_AGENT --> MERGE

    DEFAULT --> MERGE

    

    MERGE --> SETTINGS[AgentSettings 单例]

    

    subgraph "配置文件位置"

        F1[.env]

        F2[outfit_biz/config.yml]

        F3[outfit_agent/config.yml]

    end

    

    F1 --> ENV

    F2 --> YAML_BIZ

    F3 --> YAML_AGENT

配置文件的叠加规则：

1. 1. outfit_biz/config.yml 是基础配置
2. 2. outfit_agent/config.yml 在基础上叠加（deep-merge）
3. 3. 环境变量具有最高优先级，会覆盖所有配置文件中的同名配置

Sources: app/config.py, outfit_biz/src/outfit_biz/config.py

配置最佳实践

1. 安全性原则

• - 敏感信息必须使用环境变量：API 密钥、数据库密码、Secret 等敏感信息不要写入配置文件
• - .env 文件必须加入 .gitignore：防止密钥泄露到代码仓库
• - 生产环境使用随机长字符串：ADMIN_SECRET_KEY、DEV_TOKEN 等必须使用强密码

Sources: .gitignore, .env.example

2. 开发环境配置

# 复制示例文件

cp .env.example .env



# 编辑 .env 文件，填写必要的配置

# 至少需要配置：

# - OPENAI_API_KEY（LLM 调用）

# - DATABASE_URL（数据库连接）

# - REDIS_URL（Redis 缓存，可选）

开发环境可以使用 DEV_TOKEN 和 DEV_USER_ID 简化测试流程。

Sources: README.md, docs/deployment.md

3. 生产环境配置

生产环境配置建议：

1. 1. 使用环境变量而非 .env 文件：通过容器编排（Docker/K8s）注入环境变量
2. 2. 启用内容安全审核：设置 ALIYUN_CONTENT_AUDIT_ENABLED=true
3. 3. 配置监控：设置 HYPERDX_API_KEY 和 OTEL_SERVICE_NAME
4. 4. 启用 Redis 缓存：设置 LITELLM_REDIS_CACHE_ENABLED=true 降低 LLM 调用成本
5. 5. 配置合理的并发参数：根据服务器资源调整 IMG_GEN_MAX_CONCURRENCY 等参数

Sources: ecosystem.config.js, docs/deployment.md

4. 配置验证

系统启动时会自动验证配置的有效性：

• - 日志级别必须在 DEBUG/INFO/WARNING/ERROR/CRITICAL 范围内
• - 数据库连接池大小必须大于 0
• - 超时时间必须为正数
• - 无效配置会使用默认值并输出警告日志

Sources: outfit_biz/src/outfit_biz/config.py

常见配置场景

场景 1：接入新的 LLM 提供商

如果需要接入新的 LLM 提供商（如 DeepSeek），需要配置：

DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

DEEPSEEK_BASE_URL=https://api.deepseek.com

然后在 config.yml 的 model_map 中配置模型别名。

Sources: outfit_biz/src/outfit_biz/config.py

场景 2：启用 Redis 缓存

启用 Redis 缓存可以显著降低 LLM 调用延迟和成本：

REDIS_URL=redis://localhost:6379/0

LITELLM_REDIS_CACHE_ENABLED=true

LITELLM_REDIS_TTL=3600

Sources: .env.example, app/main.py

场景 3：配置企业微信客服

配置企业微信客服需要以下变量：

WECHAT_TOKEN=your_token

WECHAT_ENCODING_AES_KEY=your_aes_key

WECHAT_CORP_ID=your_corp_id

WECHAT_SECRET=your_secret

WECHAT_MINIPROGRAM_APPID=your_appid

Sources: .env.example, app/config.py

场景 4：启用内容安全审核

启用阿里云内容安全审核：

ALIYUN_AK_ID=your_access_key_id

ALIYUN_AK_SECRET=your_access_key_secret

ALIYUN_CONTENT_AUDIT_ENABLED=true

ALIYUN_TEXT_AUDIT_ENABLED=true

ALIYUN_IMAGE_AUDIT_ENABLED=true

Sources: .env.example, app/config.py

配置变量速查表

下一步

配置完成后，建议继续阅读：

• - 配置文件结构：了解 config.yml 的详细结构和 Agent/Team 配置
• - 系统架构总览：了解系统的整体架构设计
• - 环境搭建与启动：了解如何启动和运行系统

<a id="4-pei-zhi-wen-jian-jie-gou"></a>

---

配置文件结构

• - Section: 快速入门
• - Group: 配置与环境
• - Source File: 4-pei-zhi-wen-jian-jie-gou.md

outfit-agent 的配置体系由四层配置源叠加而成：环境变量、.env 文件、config.yml YAML 配置、以及 TOML/JSON/文本等结构化数据文件。本页将系统梳理项目中所有配置文件的组织方式、加载优先级和各节用途。

配置层次总览

配置加载遵循明确的优先级链——高优先级源覆盖低优先级源的同名键值，且所有配置最终汇聚为 AgentSettings 单例对象。

graph TD

    A["① 环境变量 / 系统环境"] -->|最高优先级| M["AgentSettings 单例"]

    B["② .env 文件"] -->|次高| M

    C["③ config.yml（biz层 + agent层 deep-merge）"] -->|基础| M

    D["④ TOML / JSON / 文本数据文件"] -->|运行时按需加载| R["运行时组件"]

    M --> R



    subgraph "config.yml 合并过程"

        C1["outfit_biz/config.yml"] -->|_deep_merge| C2["outfit_agent/config.yml"]

        C2 --> C

    end

Sources: app/config.py

环境变量与 .env 文件

.env.example 列出了所有可配置的环境变量，复制为 .env 后填入实际值即可。.env 已被 .gitignore 忽略，不会提交到版本库。

环境变量的值在 app.config.AgentSettings 中通过 Pydantic BaseSettings 机制自动注入，覆盖 .env 和 config.yml 中的同名字段。

Sources: .env.example, app/config.py

主配置文件 config.yml

config.yml 是 YAML 格式的主配置文件，采用分层叠加策略：outfit_biz/config.yml 作为基础层，outfit_agent/config.yml 作为覆盖层，两者通过 _deep_merge() 递归合并——agent 层的同名键覆盖 biz 层，嵌套字典则递归合并而非整体替换。

config.yml 中仅存放可序列化的标量值（bool / int / float / str），需要运行时对象（模型实例、数据库连接、工具列表、钩子函数等）的参数始终在 Python 代码中设置。

Sources: config.yml, app/config.py

顶层配置节

config.yml 的顶层键直接映射为 AgentSettings 的字段属性，按业务域分组：

Sources: config.yml, app/config.py

agent 配置节

agent 节是 config.yml 的核心嵌套结构，控制 Agent 和 Team 的行为参数。该节下分为三个子节，分别透传给不同的构造函数：

graph TD

    YAML["config.yml agent 节"] --> L["agent.leader_model"]

    YAML --> T["agent.team"]

    YAML --> M["agent.member"]

    L -->|模型 alias| Agent["app/agent.py get_model()"]

    T -->|kwargs| TeamInit["Team.__init__()"]

    M -->|kwargs| AgentInit["Agent.__init__()"]

agent.leader_model 指定主 Team 使用的模型角色别名（如 default、flash、vision），映射到 config.yml 的 model_map 节中定义的实际模型。

agent.team 控制主 Team（Leader）行为，关键参数包括：

agent.member 控制子 Agent 默认行为，参数结构与 team 一致，但独立配置。

Sources: config.yml, app/config.py

TOML 配置文件

项目使用 TOML 格式定义 Agent 团队、技能和 Canvas Agent 的结构化配置。这些文件由对应的 Loader 类在启动时加载，并支持灰度覆盖。

主 Team 配置：data/team/main.toml

定义主 Agent 的身份、工具列表、技能绑定和渠道专属工具：

graph LR

    TOML["data/team/main.toml"] --> Loader["MainTeamLoader"]

    Loader --> Config["MainTeamConfig"]

    Config --> Tools["工具列表（全渠道 + 渠道专属）"]

    Config --> Skills["技能绑定"]

    Config --> Instructions["提示词路径"]

主 Team 配置支持灰度覆盖——通过 get_gray_config_service().get_toml_overlay() 可为特定用户提供不同的 TOML 内容。

Sources: data/team/main.toml, app/agents/team_loader.py

Canvas Agent 配置：data/team/canvas.toml

定义小程序 Canvas 专用 Agent 的参数：

Sources: data/team/canvas.toml

技能配置：data/skills/*/skill.toml

每个技能目录下包含一个 skill.toml 文件，定义技能的元数据、工具绑定和提示词路径：

graph TD

    subgraph "data/skills/"

        S1["nuva-soul/skill.toml"]

        S2["style-advisor-skill/skill.toml"]

        S3["buyer-wardrobe-skill/skill.toml"]

        S4["outfit-output-skill/skill.toml"]

    end

    S1 & S2 & S3 & S4 --> Registry["SkillRegistry"]

    Registry --> Bundle["SkillBundle"]

    Bundle --> Agent["运行时 Agent"]

技能 TOML 的标准字段：

当前项目注册了四个技能：

技能配置同样支持灰度覆盖，灰度范围以 data/skills/<id>/skill.toml 为 scope。

Sources: data/skills/nuva-soul/skill.toml, data/skills/buyer-wardrobe-skill/skill.toml, data/skills/outfit-output-skill/skill.toml, data/skills/example/skill.toml, app/skills/loader.py

工具描述文件：data/tools/descriptions.toml

工具描述文件以 TOML 格式集中管理所有工具的自然语言描述。这些描述会在运行时注入到 agno.tools.Function 对象上，作为 LLM 决策工具选择的核心依据。

文件结构为 [tools] 节下的键值对，键为工具名，值为多行描述文本。描述中包含使用场景、参数约束、返回值格式和与其他工具的边界说明。

该文件同样支持灰度覆盖——通过 user_id 参数加载用户专属的描述变体，materialize_tools_for_user() 函数负责合并基础描述与用户覆盖。

Sources: data/tools/descriptions.toml, app/tools/factory.py

提示词文件：data/prompts/

提示词文件以 .txt 纯文本格式存放在 data/prompts/ 目录下，按业务域组织子目录，所有文件均提供中英文双版本（.zh.txt / .en.txt）：

提示词文件支持动态占位符：{category_taxonomy_yaml} 和 {category_taxonomy} 会在加载时被替换为实际的品类分类数据。

提示词加载还支持渠道覆盖——data/channels/<channel_name>/ 下的同名文件会覆盖全局提示词，实现不同渠道的差异化指令。加载顺序为：渠道覆盖 → Agent 本地 → Biz 默认。

Sources: data/prompts/agents/main_team.zh.txt, app/i18n.py, app/channel_prompts.py

国际化数据：data/i18n/

JSON 格式的国际化字符串文件，提供运行时文本的中英文对照：

这些文件通过 app.i18n.get_text() 按键路径读取，语言参数默认为 "zh"。

Sources: data/i18n/system.zh.json, app/i18n.py

配置加载机制

配置加载的核心在 app.config 模块中实现，关键流程如下：

flowchart TD

    A["启动 / get_settings()"] --> B["get_biz_settings()"]

    B --> C["加载 outfit_biz/config.yml"]

    B --> D["加载 .env"]

    B --> E["读取环境变量"]

    C --> F["_deep_merge(biz_yml, agent_yml)"]

    G["加载 outfit_agent/config.yml"] --> F

    F --> H["_merged_yml 全局字典"]

    H --> I["AgentSettings(**biz.model_dump())"]

    I --> J["@lru_cache 单例"]

合并策略：

1. 1. outfit_biz/config.yml 提供基础配置
2. 2. outfit_agent/config.yml 递归覆盖——同名标量直接替换，嵌套字典递归合并
3. 3. 环境变量和 .env 通过 Pydantic BaseSettings 注入，优先级最高
4. 4. AgentSettings 继承 BizSettings，添加 Agent 专属字段

AgentSettings 中的属性访问器（如 team_params、member_params、leader_model）直接从 _merged_yml 读取对应的嵌套字典，透传给 Team.__init__() 和 Agent.__init__()。

Sources: app/config.py

灰度配置

项目支持用户级别的配置灰度覆盖，通过 outfit_biz.gray.GrayConfigService 实现。灰度覆盖可作用于：

• - TOML 配置：为特定用户提供不同的 main.toml 或 skill.toml 内容
• - 工具描述：为特定用户提供定制化的工具描述文本
• - 提示词：为特定用户提供差异化的提示词

灰度签名基于 user_id 和 scope 构建，相同签名的用户共享同一份配置变体缓存。未命中灰度规则的用户使用 base 配置。

Sources: app/agents/team_loader.py, app/skills/loader.py, app/tools/factory.py

配置文件全景图

以下树形结构展示项目中所有配置相关文件的组织方式：

outfit_agent/

├── .env.example                      # 环境变量模板

├── config.yml                        # Agent 层 YAML 主配置（叠加在 biz 层之上）

│

├── data/

│   ├── team/

│   │   ├── main.toml                 # 主 Team 配置（工具、技能、渠道工具）

│   │   └── canvas.toml               # Canvas Agent 配置

│   │

│   ├── skills/

│   │   ├── nuva-soul/skill.toml      # 人格技能

│   │   ├── style-advisor-skill/      # 穿搭建议技能

│   │   ├── buyer-wardrobe-skill/     # 搜品衣橱技能

│   │   └── outfit-output-skill/      # 视觉输出技能

│   │

│   ├── tools/

│   │   └── descriptions.toml         # 工具描述集中管理

│   │

│   ├── prompts/                      # 提示词文件（.txt，中英双版本）

│   │   ├── agents/                   # Agent 系统指令

│   │   ├── skills/                   # 技能指令

│   │   ├── memory/                   # 记忆/洞察提示词

│   │   ├── workflows/                # 工作流提示词

│   │   ├── partials/                 # 可复用片段

│   │   └── system/                   # 系统级提示词

│   │

│   ├── i18n/                         # 国际化 JSON 字符串

│   │   ├── system.zh.json

│   │   ├── system.en.json

│   │   └── ...

│   │

│   └── channels/                     # 渠道覆盖文件

│       ├── wechat_kefu/              # 微信客服渠道覆盖

│       └── wechat_miniapp/           # 小程序渠道覆盖

│

└── app/

    ├── config.py                     # AgentSettings 定义与配置加载

    ├── i18n.py                       # 国际化加载与搜索路径注册

    ├── channel_prompts.py            # 渠道提示词覆盖逻辑

    ├── agents/team_loader.py         # 主 Team TOML 配置加载器

    ├── skills/loader.py              # 技能 TOML 注册表

    └── tools/factory.py              # 工具描述加载与灰度合并

下一步

• - 如需了解环境变量的详细分类和用途，请参阅 环境变量配置
• - 如需了解配置如何驱动系统整体运行，请参阅 系统架构总览
• - 如需了解技能加载与管理的完整流程，请参阅 技能加载与管理
• - 如需了解工具描述管理与灰度机制，请参阅 工具描述管理
• - 如需了解提示词管理与国际化，请参阅 提示词管理与国际化

<a id="5-xi-tong-jia-gou-zong-lan"></a>

---

系统架构总览

• - Section: 架构与核心概念
• - Source File: 5-xi-tong-jia-gou-zong-lan.md

本页是 Outfit Agent 系统的全局架构导览，面向首次接触代码库的开发者。文章从分层架构、请求生命周期、核心子系统三个维度，帮助你建立对整个系统的宏观认知，并指引你进入各专题的深度阅读。

技术栈概览

Outfit Agent 是一个基于 FastAPI + Agno Agent 框架 构建的智能穿搭顾问 REST API 服务。下表汇总了系统的核心技术选型与职责：

Sources: README.md, app/main.py, app/agentos.py, app/observability.py

分层架构

整个系统遵循清晰的分层结构，自上而下依次为：入口路由层 → 消息处理层 → Agent 层 → 工具层 → 服务层 → 数据/记忆层。

graph TB

    subgraph 入口路由层

        A1[微信客服路由<br/>routers/wechat.py]

        A2[小程序路由<br/>routers/miniapp/*]

        A3[REST Chat 路由<br/>routers/chat.py]

        A4[Admin API<br/>routers/admin/*]

        A5[Canvas WS 路由<br/>routers/miniapp/canvas_ws.py]

    end



    subgraph 消息处理层

        B1[消息入站归一化<br/>messaging/inbound.py]

        B2[会话分发状态机<br/>messaging/session_dispatcher.py]

        B3[Agent 执行服务<br/>messaging/agent_service.py]

        B4[回复状态管理<br/>messaging/reply_state.py]

        B5[快速响应<br/>fast_response.py]

    end



    subgraph Agent 层

        C1[主 Agent 单例<br/>agent.py]

        C2[Canvas Agent 单例<br/>canvas/agent.py]

        C3[Team 配置加载<br/>agents/team_loader.py]

        C4[技能注册表<br/>skills/loader.py]

        C5[Dialog State<br/>dialog_state.py]

    end



    subgraph 工具层

        D1[工具工厂<br/>tools/factory.py]

        D2[工具注册表<br/>tools/__init__.py]

        D3[ToolContext<br/>tools/context.py]

        D4[业务工具集<br/>tools/*.py]

    end



    subgraph 服务层

        E1[穿搭工作流<br/>workflows/outfit_generation.py]

        E2[图片生成<br/>img_gen_handler.py]

        E3[单品初始化<br/>item_init_handler.py]

        E4[图片描述<br/>image_caption_handler.py]

        E5[Outfit 服务<br/>services/outfit_*.py]

    end



    subgraph 数据与记忆层

        F1[(PostgreSQL)]

        F2[(Redis)]

        F3[Mem0 向量记忆<br/>memory/client.py]

        F4[用户洞察<br/>insights/service.py]

    end



    A1 & A2 & A3 & A5 --> B1

    A4 -.-> |admin 直连| F1

    B1 --> B2 --> B3

    B3 --> C1 & C2

    B4 & B5 -.-> B3

    C1 --> C3 --> C4

    C1 & C2 --> C5

    C1 & C2 --> D1 --> D2 --> D4

    D1 --> D3

    D4 --> E1 & E2 & E3 & E5

    E2 & E3 & E4 --> F2

    C1 & C2 --> F1

    C1 --> F3

    F4 --> F3

    E1 --> F1

Sources: app/main.py, app/tools/__init__.py, app/messaging/session_dispatcher.py

应用启动流程

系统启动时通过 _build_app() 函数完成一系列初始化，最终返回一个融合了业务路由与 AgentOS 路由的 ASGI 应用。启动顺序如下：

sequenceDiagram

    participant main as main.py

    participant agent as Agent 初始化

    participant app as FastAPI

    participant agentos as AgentOS

    participant lifespan as Lifespan



    main->>main: load_dotenv(.env)

    main->>main: _init_httpx_client()

    main->>agent: init_agent() — 构建主 Agent 单例

    main->>agent: init_canvas_agent() — 构建 Canvas Agent 单例

    main->>main: init_hyperdx_monitoring()

    main->>app: 创建 FastAPI 实例 + 注册路由

    main->>agentos: get_enriched_app(base_app, lifespan)

    Note over agentos: AgentOS 挂载 tracing / scheduler / MCP



    Note over lifespan: ← 进入 lifespan

    lifespan->>lifespan: DB schema migration (users/outfits/items)

    lifespan->>lifespan: ensure_gray_config_schema()

    lifespan->>lifespan: ensure_insight_schema()

    lifespan->>lifespan: ensure_message_access_control_schema()

    lifespan->>lifespan: init_memory() — Mem0 AsyncMemory 单例

    lifespan->>lifespan: _init_litellm_redis_cache()

    lifespan->>lifespan: recover_interrupted_workflows()

    lifespan->>lifespan: ImgGenHandler.start()

    lifespan->>lifespan: ImageCaptionHandler.start()

    lifespan->>lifespan: ItemProcessingHandler.start()

    Note over lifespan: ← yield — 服务就绪

Sources: app/main.py, app/main.py, app/agentos.py

双 Agent 架构

系统运行时存在两个独立的 Agent 单例，各自服务不同的交互场景：

两者共享同一个 PostgreSQL 数据库（schema=agno）和同一套工具注册表，但拥有独立的 prompt 指令、工具列表和 session state 隔离。Canvas Agent 通过 canvas/bridge.py 将编辑摘要桥接回主对话上下文。

Sources: app/agent.py, app/canvas/agent.py, data/team/main.toml, app/canvas/bridge.py

消息处理生命周期

一条用户消息从进入系统到 Agent 回复，经历以下核心阶段：

flowchart TD

    A[用户消息到达] --> B{渠道识别}

    B -->|微信客服| C[WXBizMsgCrypt 解密]

    B -->|小程序| D[Token 鉴权 / WebSocket]

    B -->|REST API| E[LoginRequired 校验]



    C & D & E --> F[消息入站归一化<br/>messaging/inbound.py]

    F --> G[入站快捷回复检查<br/>inbound_shortcuts.py]

    G -->|命中快捷回复| H[直接返回固定回复]

    G -->|未命中| I[内容审核<br/>aliyun_text_audit.py]



    I --> J{会话分发状态机<br/>session_dispatcher.py}

    J -->|图片消息| K[标记 image_in_flight]

    J -->|文本消息| L[文本任务调度]



    K --> M[图片处理管线<br/>image_pipeline.py]

    M --> N[图片描述 / 存储]

    N --> O[合并到 Agent 输入]



    L --> P{检查任务冲突}

    P -->|可取消| Q[合并待处理消息]

    P -->|需等待| R[入 Redis 队列排队]



    Q & O --> S[Agent Turn 执行<br/>agent_service.py]

    S --> T[build_tool_ctx → ToolContext]

    T --> U[Agent.arun]

    U --> V[LLM 调用 + Tool 调用链]

    V --> W[回复过滤<br/>display_filters.py]

    W --> X[持久化回复消息]

    X --> Y[渠道投递<br/>send_text_message]

Sources: app/routers/wechat.py, app/routers/chat.py, app/messaging/session_dispatcher.py, app/messaging/agent_service.py

工具系统

工具（Tool）是 Agent 与外部世界交互的桥梁。系统通过 工厂模式 + 注册表模式 管理所有工具：

工具工厂 (tools/factory.py) 负责：将普通 Python 函数包装为 Agno Function 对象；在运行时根据用户 ID 动态加载 TOML 格式的工具描述覆盖；以及注入工具门控逻辑（Dialog State 检查、渠道权限检查、技能绑定检查）。

工具注册表 (tools/__init__.py) 是一个全局 dict[str, Function]，包含以下工具类别：

ToolContext (tools/context.py) 是一个会话级别的上下文对象，存储在 Agno session_state["tool_ctx"] 中，包含 user_id、lang、session_id、avatar_path、channel 等信息，所有业务工具通过 _get_ctx(run_context) 提取，避免依赖 LLM 传递用户身份参数。

Sources: app/tools/factory.py, app/tools/__init__.py, app/tools/context.py

技能系统

技能（Skill）是组织工具和指令的业务单元。每个技能由 data/skills/<id>/skill.toml 声明，包含：绑定的工具列表、渠道专属工具、激活条件、参考文档和脚本。

SkillRegistry (skills/loader.py) 在启动时扫描 data/skills/ 目录，构建 RegisteredSkill 对象。主 Agent 的 data/team/main.toml 通过 skills 列表引用技能 ID：

skills = [

    "nuva-soul",

    "style-advisor-skill",

    "buyer-wardrobe-skill",

    "outfit-output-skill",

]

技能的工具通过 门控机制 控制可见性：当 active_skill 字段匹配技能 ID 时，该技能的工具才对 Agent 可用。这确保 Agent 在不同 SOP 阶段只暴露相关工具，降低 LLM 决策复杂度。

Sources: app/skills/loader.py, data/team/main.toml

Dialog State 与 SOP 流程

Dialog State 是嵌入在 Agno session_state["dialog_state"] 中的有限状态机，驱动整个对话的 SOP（标准操作流程）推进。其核心字段包括：

Agent 通过 update_dialog_state 工具推进状态，get_dialog_state 工具读取状态。工具门控函数 is_tool_allowed() 根据 active_skill 决定工具是否可用。

Sources: app/dialog_state.py

会话分发状态机

当用户快速连续发送多条消息时，会话分发器 (messaging/session_dispatcher.py) 通过模块级状态字典确保消息的顺序处理：

• - _text_tasks：每个 session 最多一个正在运行的 Agent 文本任务
• - _image_processing：图片消息处理独占锁——图片处理期间阻塞所有后续消息
• - 消息合并：可取消的待处理文本消息会合并到下一条消息中（debounce 行为）
• - 队列排空：任务完成后自动调用 drain_queue() 处理队列中的下一条消息

渠道特定路由（如微信客服）只需提供 on_text 和 prepare_image 两个回调，即可接入统一的分发状态机。

Sources: app/messaging/session_dispatcher.py

穿搭工作流

穿搭生成采用 异步工作流架构，通过 Agno Workflow 实现多步骤编排：

flowchart LR

    A[prepare_context] --> B[ideate_outfits]

    B --> C[select_items]

    C --> D[generate_images]

    D --> E[dispatch_results]



    style A fill:#e8f5e9

    style B fill:#e3f2fd

    style C fill:#fff3e0

    style D fill:#fce4ec

    style E fill:#f3e5f5

工作流通过 OutfitWorkflowService 以 asyncio.create_task 在后台运行。步骤 B 由 OutfitIdeationService 调用 LLM 生成穿搭创意；步骤 C 由 OutfitItemSelectionService 从用户衣柜中匹配单品；步骤 D 将任务入 Redis 队列，由后台 ImgGenHandler 消费并调用图片生成 API。

背景任务处理器是一个统一模式：Redis 队列负责任务调度，数据库 reconcile 保障幂等性，processing lock 防止重复消费。系统中有三个此类处理器：

Sources: app/workflows/outfit_generation.py, app/services/outfit_workflow_service.py, app/img_gen_handler.py, app/main.py

记忆与洞察

Mem0 记忆 (memory/) 提供基于 pgvector 的用户长期记忆。系统在 Agent 每轮对话前检索相关记忆（search_user_memory），在对话后提取并持久化新记忆（add_conversation_memory）。配置项 mem0_realtime_analysis_enabled 控制是否启用逐轮实时记忆提取。

用户洞察 (insights/) 是离线分析管线，通过 Agno Cron Scheduler 每日执行，分析用户的穿搭数据（outfit）、单品数据（item）和聊天数据（chat），生成结构化洞察并写入 user_insights 表。洞察结果可同步回 Mem0 记忆层。

Sources: app/memory/client.py, app/memory/__init__.py, app/insights/service.py

渠道抽象

系统支持多个接入渠道，通过 渠道名 统一抽象：

渠道差异通过 data/team/main.toml 中的 [channels.*] 段管理工具可见性，以及 data/i18n/ 和 data/prompts/ 下的文件覆盖实现 prompt 国际化。ToolContext.channel 字段贯穿整个调用链，使工具层能够感知当前渠道并做出差异化行为。

Sources: data/team/main.toml, app/tools/factory.py, app/routers/wechat.py

配置与灰度

配置采用 双层 YAML 合并 策略：outfit_biz/config.yml 作为基础配置，outfit_agent/config.yml 作为覆盖层，通过深度递归合并生成最终配置。AgentSettings 继承自 BizSettings，增加了 Agent 专有字段（如微信配置、Mem0 参数、后台处理器并发数等）。

灰度系统 支持按用户 ID 对配置文件（包括 team TOML、skill TOML、tool descriptions TOML）施加覆盖，实现 A/B 测试。各加载器（MainTeamLoader、SkillRegistry、_load_tool_descriptions）均内置了灰度签名计算和缓存。

Sources: app/config.py

管理与可观测性

• - Admin API (routers/admin/)：提供 20+ 个管理端点，涵盖用户管理、Agent 配置热更新、技能管理、灰度规则、Prompt 编辑、链路追踪查看、Mem0 记忆管理等
• - Dashboard：SPA 应用，构建产物挂载在 /dashboard/ 路径下
• - Live Runs (live_runs.py)：将当前正在执行的 Agent Run 实时写入 Redis，供 Dashboard 监控
• - HyperDX：通过 OpenTelemetry OTLP 协议将日志导出到 HyperDX 平台
• - AgentOS Tracing：通过 OpenTelemetry + DatabaseSpanExporter 将 Agent 链路追踪数据写入 PostgreSQL（schema=agno）

Sources: app/live_runs.py, app/observability.py, app/agentos.py

项目目录结构

outfit_agent/

├── app/                        # 应用核心代码

│   ├── main.py                 # FastAPI 入口 + lifespan 管理

│   ├── agent.py                # 主 Agent 单例构建与运行

│   ├── agentos.py              # AgentOS 集成（tracing/scheduler/MCP）

│   ├── config.py               # 配置加载与合并

│   ├── dialog_state.py         # Dialog State 有限状态机

│   ├── session.py              # 会话 ID 管理

│   ├── i18n.py                 # 国际化桥接

│   ├── fast_response.py        # 快速响应机制

│   ├── agents/                 # Agent 配置加载（team_loader, profile_hook）

│   ├── canvas/                 # Canvas Agent（agent, bridge, context, service）

│   ├── insights/               # 离线用户洞察管线

│   ├── memory/                 # Mem0 记忆集成

│   ├── messaging/              # 消息处理（inbound, dispatcher, agent_service）

│   ├── routers/                # HTTP/WebSocket 路由

│   │   ├── chat.py             # REST Chat API

│   │   ├── wechat.py           # 微信客服回调

│   │   ├── miniapp/            # 小程序路由集

│   │   └── admin/              # Admin API 端点集

│   ├── services/               # 业务服务层

│   ├── skills/                 # 技能加载与注册

│   ├── tools/                  # 工具工厂、注册表与业务工具

│   └── workflows/              # 穿搭生成工作流

├── data/                       # 非代码资源

│   ├── team/                   # Agent/Team TOML 配置

│   ├── skills/                 # 技能定义目录

│   ├── prompts/                # Prompt 模板

│   ├── i18n/                   # 国际化 JSON

│   └── tools/                  # 工具描述 TOML

├── dashboard/                  # 管理面板 SPA

├── tests/                      # 测试套件

├── scripts/                    # 运维脚本

├── config.yml                  # Agent 层配置覆盖

└── pyproject.toml              # 项目元数据与依赖

Sources: 项目目录结构, README.md

推荐阅读路径

基于以上架构全景，建议按以下顺序深入各专题：

1. 1. 单主 Agent 架构 — 理解 Agent 单例构建、模型解析、运行时变体机制
2. 2. 技能加载与管理 — 掌握 Skill 注册表、TOML 配置与灰度机制
3. 3. Dialog State 设计 — 理解 SOP 流程如何驱动对话状态推进
4. 4. 工具注册与工厂 — 深入工具包装、门控、描述管理
5. 5. 消息入站处理流程 — 从用户消息到 Agent 调用的完整链路
6. 6. 穿搭工作流服务 — 异步工作流编排与图片生成管线

<a id="6-dan-zhu-agent-jia-gou"></a>

---

单主 Agent 架构

• - Section: 架构与核心概念
• - Group: Agent 与技能系统
• - Source File: 6-dan-zhu-agent-jia-gou.md

本文档详细介绍了 Outfit Agent 项目采用的单主 Agent 架构设计。该架构以 agno 框架为基础，采用单一 Agent 作为核心业务编排者，通过动态技能加载和显式 SOP 状态推进，实现了高效、可控的智能穿搭顾问系统。

架构设计原则

单主 Agent 架构的核心设计理念源于对传统多 Agent 协作模式的反思与优化。在早期架构中，系统采用"主 Team 协调多个子 agent"的模式，但这种模式存在几个关键问题：场景容易被整体交给某个子 agent 处理、系统容易自动跑完整长链路而缺乏中间确认点、会话状态管理不够显式。

基于这些挑战，当前架构遵循以下核心原则：一轮一个主目标——每轮交互只处理一个主要意图；一步一个动作——避免在单次调用中执行过多操作；关键动作前先确认——在重要决策点暂停等待用户反馈；当前动作完成后及时收口——确保每次交互都有明确的结束状态。这些原则共同保证了系统行为的可预测性和用户体验的可控性。

Sources: app/agent.py, docs/agno_agents.md, docs/agent_refact.md

核心组件架构

单主 Agent 架构由三个核心组件构成：主 Agent 实例、技能注册系统和工具工厂系统。这三个组件协同工作，形成了完整的业务处理流程。

主 Agent 实例是整个系统的唯一业务编排者，由 app/agent.py 中的 init_agent() 函数初始化并缓存为全局单例。该 Agent 承担意图识别、技能加载决策、SOP 阶段推进和最终回复生成等核心职责。Agent 实例通过 _build_agent() 函数构建，该函数整合了模型配置、团队配置、技能包和工具列表，最终创建一个可复用的 Agent 对象。

技能注册系统由 app/skills/loader.py 中的 SkillRegistry 类实现。该系统负责扫描 data/skills/ 目录下的技能定义文件（skill.toml），构建技能对象并注册到全局注册表中。每个技能包含激活条件、指令提示词、工具列表和访问控制规则等元数据。技能的激活采用"按需加载"模式，只有当主 Agent 识别到特定意图时才会调用相应的技能指令。

工具工厂系统由 app/tools/factory.py 中的 build_tool() 函数和 TOOL_REGISTRY 字典共同构成。该系统负责将业务函数包装为 agno 可识别的 Function 对象，并注入必要的上下文（用户信息、会话状态、渠道数据等）。工具描述从 data/tools/descriptions.toml 文件加载，支持灰度配置覆盖。

Sources: app/agent.py, app/skills/loader.py, app/tools/factory.py

单例初始化与生命周期管理

主 Agent 的初始化遵循严格的启动序列，确保所有依赖组件在使用前已正确配置。整个初始化过程由 init_agent() 函数协调，该函数在应用启动时被调用一次。

sequenceDiagram

    participant Main as app/main.py

    participant Agent as app/agent.py

    participant Tools as app/tools

    participant Skills as app/skills

    participant Loader as team_loader.py



    Main->>Agent: init_agent()

    Agent->>Agent: _make_agent_db()

    Agent->>Agent: _make_agent_sync_engine()

    Agent->>Tools: init_tool_registry()

    Agent->>Skills: init_skill_registry()

    Agent->>Loader: get_main_team_loader().reload()

    Agent->>Agent: _resolve_model_runtime(leader_model)

    Agent->>Agent: _build_agent(leader_runtime)

    Agent-->>Main: Agent singleton ready

初始化完成后，Agent 实例被缓存在 _agent 全局变量中，并通过 get_agent() 函数提供给其他模块使用。该函数支持基于用户 ID 的灰度变体缓存，确保不同用户群体可以使用不同的 Agent 配置而互不干扰。

Agent 的生命周期与应用进程绑定，支持热重载功能。当配置文件或技能定义发生变化时，可以通过调用 reload_agents() 函数重新构建 Agent 实例，无需重启整个应用。热重载过程会依次重新初始化工具注册表、技能注册表和团队加载器，然后重新构建 Agent 实例。

Sources: app/agent.py, app/main.py

配置驱动的团队定义

主 Agent 的行为由 data/team/main.toml 配置文件定义，该文件采用 TOML 格式，清晰地声明了 Agent 的工具、技能和指令集。这种配置驱动的设计使得系统行为可以在不修改代码的情况下进行调整。

配置文件的主要结构包括：

• - id 和 name：Agent 的唯一标识和显示名称
• - tools：主 Agent 常驻可用的基础工具列表
• - skills：主 Agent 可用的技能列表
• - instructions：多语言指令提示词路径
• - channels：渠道专属工具配置

例如，当前配置声明了 run_outfit_flow、update_user_profile、new_session 等基础工具，以及 nuva-soul、style-advisor-skill、buyer-wardrobe-skill、outfit-output-skill 等核心技能。渠道配置允许为特定渠道（如微信客服）声明专属工具，实现场景化的功能控制。

配置加载由 MainTeamLoader 类负责，该类支持灰度配置覆盖。当指定了用户 ID 时，加载器会查询灰度配置服务，根据用户所属的灰度组应用不同的配置覆盖，从而实现 A/B 测试和渐进式发布。

Sources: data/team/main.toml, app/agents/team_loader.py

技能激活与工具门控机制

技能激活是单主 Agent 架构的核心动态行为。当主 Agent 识别到用户意图需要特定技能时，会通过 get_skill_instructions 工具调用加载相应的技能指令。这个过程同时会更新对话状态中的 active_skill 字段，触发工具门控机制。

工具门控机制由 app/dialog_state.py 中的 is_tool_allowed() 函数实现。该函数检查当前工具是否属于某个技能，以及该技能是否已被激活。如果工具属于某个技能但该技能未被激活，工具调用将被拒绝并返回错误信息。这种机制确保了工具只能在正确的上下文中使用，避免了越权操作。

flowchart TD

    A[用户消息] --> B[主Agent意图识别]

    B --> C{需要特定技能?}

    C -->|是| D[调用get_skill_instructions]

    D --> E[更新dialog_state.active_skill]

    E --> F[技能指令加载完成]

    C -->|否| G[使用主Agent内建能力]

    F --> H[工具门控检查]

    G --> H

    H --> I{工具是否允许?}

    I -->|是| J[执行工具]

    I -->|否| K[返回错误信息]

    J --> L[更新对话状态]

    L --> M[生成用户回复]

技能激活还支持"always instruction"模式，某些技能的核心指令会被注入到主 Agent 的基础指令中，确保这些技能的行为准则始终生效。例如，nuva-soul 技能的人格定义指令就采用这种模式，保证主 Agent 在任何交互中都保持一致的人格特征。

Sources: app/skills/access_tools.py, app/dialog_state.py, app/tools/factory.py

会话状态与对话管理

会话状态管理是单主 Agent 架构保持对话连贯性的关键。系统采用分层状态设计：session_state 存储会话级数据，dialog_state 存储对话流程状态，ToolContext 存储请求级上下文。

dialog_state 是对话流程管理的核心数据结构，由 app/dialog_state.py 定义。它包含以下关键字段：

• - active_skill：当前激活的技能 ID
• - loaded_skills：已加载的技能列表
• - sop_stage：当前 SOP 阶段
• - active_scene：当前活跃场景
• - outfit_task：穿搭任务配置
• - outfit_status：穿搭生成状态
• - pending_confirmation：待用户确认的内容

这些字段共同构成了对话流程的状态机，主 Agent 根据这些状态决定下一步行为。例如，当 sop_stage 为 await_outfit_confirmation 时，Agent 会等待用户确认而不是继续执行下一步操作。

ToolContext 是请求级上下文，包含用户 ID、语言偏好、渠道信息、头像路径等数据。它通过 set_session_tool_ctx() 函数注入到 session_state 中，并在工具调用时通过 _get_ctx() 函数提取。这种设计确保了每个工具调用都能获得正确的上下文信息。

Sources: app/dialog_state.py, app/tools/context.py

消息处理与并发控制

消息处理采用会话级并发控制机制，确保每个会话同时只有一个 Agent 调用在执行。该机制由 app/messaging/session_dispatcher.py 实现，采用状态机模式管理文本和图片消息的处理顺序。

状态机的核心状态包括：

• - _text_tasks：当前正在执行的文本处理任务
• - _image_processing：正在处理的图片任务
• - _queue_draining：正在排空队列的会话

当用户发送文本消息时，系统首先检查是否有图片正在处理。如果有，文本消息会被加入待处理队列；如果没有，则检查是否有正在执行的文本任务。如果存在文本任务且新消息可以合并（防抖行为），则取消当前任务并将新消息合并到队列中；否则直接创建新任务执行。

图片消息的处理具有更高的优先级。一旦图片开始处理，所有后续消息（包括文本）都会被阻塞，直到图片处理完成。这种设计确保了图片处理流程的完整性，避免了图片处理过程中被其他消息干扰。

Sources: app/messaging/session_dispatcher.py

内存与上下文增强

系统集成了 Mem0 记忆模块，在每次 Agent 调用前检索相关记忆，为对话提供个性化上下文。记忆检索由 app/memory 模块的 search_user_memory() 函数实现，该函数基于语义相似度搜索用户的历史对话和记忆片段。

记忆上下文的注入遵循"读取前、写入后"的模式：

1. 1. 读取前：在构建 Agent 输入前，先检索与当前消息相关的记忆
2. 2. 构建输入：将记忆上下文、运行时上下文和用户消息组合成完整的输入
3. 3. 写入后：在获得 Agent 回复后，异步地将对话记录添加到记忆中

这种设计确保了记忆操作不会阻塞主处理流程，同时为 Agent 提供了丰富的历史上下文。记忆上下文的长度受配置限制，避免过长的记忆片段影响 Agent 的处理效率。

Sources: app/agent.py, app/memory/__init__.py

快速响应与用户体验优化

为了提升用户体验，系统实现了快速响应机制，在 Agent 处理耗时操作时向用户发送即时反馈。该机制由 app/fast_response.py 模块实现，采用"工具调用响应优先"策略。

快速响应的工作流程：

1. 1. 在 Agent 调用开始时，注册快速响应请求
2. 2. 启动异步任务监听第一个工具调用响应
3. 3. 如果在指定超时时间内收到工具调用响应，立即将该响应发送给用户
4. 4. 如果超时，则发送预设的 fallback 文本

这种机制特别适用于需要调用外部 API 的场景，如天气查询、图片生成等。用户可以在 Agent 完成完整处理前就获得初步反馈，显著降低了感知延迟。

快速响应还支持渠道特定的格式化。例如，微信客服渠道会解析菜单标记语法，将特定格式的文本转换为小程序卡片消息，提供更丰富的交互体验。

Sources: app/fast_response.py, app/agent.py

AgentOS 集成与监控

主 Agent 通过 app/agentos.py 模块集成到 Agno AgentOS 系统，获得完整的运维支持能力。AgentOS 提供了三大核心特性：追踪（Tracing）、调度（Scheduler）和 MCP Server。

追踪系统基于 OpenTelemetry 标准，将 Agent 的执行轨迹、工具调用和性能指标记录到 PostgreSQL 数据库。追踪数据存储在 agno_traces 和 agno_spans 表中，支持通过 Dashboard 或 Admin API 进行查询和分析。

调度系统支持 Cron 表达式定义的定时任务，可用于定期执行数据清理、统计计算等后台操作。调度任务通过 /v1/schedules API 进行管理。

MCP Server 提供了标准化的服务端点，允许外部系统通过 MCP 协议与 Agent 进行交互。

AgentOS 采用"base_app"模式挂载到现有的 FastAPI 应用上，保留所有业务路由的同时，添加 AgentOS 的管理路由。路由冲突处理策略为"preserve_base_app"，确保业务路由优先级高于 AgentOS 路由。

Sources: app/agentos.py, app/main.py

热重载与配置更新

系统支持运行时热重载，允许在不重启服务的情况下更新 Agent 配置。热重载通过 reload_agents() 函数触发，该函数执行以下步骤：

1. 1. 重新初始化工具注册表，加载最新的工具定义
2. 2. 重新初始化技能注册表，扫描技能目录的变更
3. 3. 重新加载团队配置文件
4. 4. 清除 Agent 变体缓存
5. 5. 使用新配置重新构建 Agent 实例

热重载过程是原子性的，在重新构建完成前，现有 Agent 实例继续处理请求。新实例构建完成后，通过全局变量替换实现无缝切换。这种设计确保了配置更新期间的服务连续性。

热重载还支持灰度配置的更新。当灰度配置发生变化时，相关用户的 Agent 变体会被重新构建，确保灰度策略的及时生效。

Sources: app/agent.py

错误处理与容错机制

系统实现了多层错误处理机制，确保在各种异常情况下都能提供合理的响应。

模型调用错误：Agent 配置了重试机制，当 LLM 调用失败时会自动重试。重试次数和超时时间通过配置参数控制。

工具执行错误：工具包装层捕获所有工具执行异常，记录详细的错误日志，并向 Agent 返回结构化的错误信息。Agent 可以根据错误类型决定是重试、跳过还是终止当前任务。

会话状态异常：当会话状态加载或持久化失败时，系统会降级到默认状态继续处理，确保对话不会因为状态异常而中断。

内存操作错误：记忆检索和写入操作采用异步执行模式，失败时只记录日志而不影响主处理流程。

所有错误都会记录到追踪系统中，支持后续的问题诊断和性能优化。

Sources: app/agent.py, app/tools/factory.py

性能优化策略

单主 Agent 架构采用了多项性能优化措施：

模型实例缓存：通过 _models 字典缓存已构建的模型实例，避免重复创建。模型实例的缓存键基于配置哈希生成，确保配置变更时能创建新的实例。

工具描述缓存：工具描述文件加载后会被缓存，支持用户级别的覆盖。缓存采用 LRU 策略，自动淘汰不常用的条目。

Agent 变体缓存：不同用户群体的 Agent 变体会被缓存，避免为每个请求都重新构建。缓存键包含模型配置和灰度签名的哈希值。

异步操作优化：记忆写入、指标上报等非关键路径操作采用异步执行，不阻塞主处理流程。

连接池管理：全局 HTTP 客户端配置了连接池，支持高并发场景下的连接复用。

这些优化措施共同确保了系统在高并发场景下的响应性能和资源利用率。

<a id="7-ji-neng-jia-zai-yu-guan-li"></a>

---

技能加载与管理

• - Section: 架构与核心概念
• - Group: Agent 与技能系统
• - Source File: 7-ji-neng-jia-zai-yu-guan-li.md

本文档详细说明 Outfit Agent 系统中技能（Skill）的定义、加载、注册与运行时管理机制。技能是系统实现"单主 Agent + 动态 Skills"架构的核心抽象，它将业务能力模块化，使主 Agent 能够按需加载不同的业务场景能力。

技能系统概述

Outfit Agent 采用单主 Agent + 动态技能架构，技能是该架构的核心组成部分。主 Agent 负责意图识别、SOP 推进和对话状态管理，而具体的业务能力（如搜品、穿搭输出、人格表达）则封装在独立的技能模块中。这种设计实现了业务逻辑与核心调度的解耦，使得新增业务能力只需添加技能定义文件，无需修改主 Agent 代码。

技能系统的核心职责包括：

1. 1. 业务能力封装：每个技能定义一组工具集合、指令提示词和激活条件
2. 2. 工具门控：技能拥有的工具只有在技能被激活后才能被调用
3. 3. 渠道适配：支持为不同渠道（如微信客服、小程序）配置不同的工具集
4. 4. 灰度发布：支持按用户 ID 对技能配置进行灰度覆盖

Sources: app/skills/loader.py, docs/agent_skill流程划分.md

技能定义文件结构

技能定义文件存放在 data/skills/ 目录下，支持两种布局方式：

目录式布局（推荐）：

data/skills/<skill-id>/

├── skill.toml          # 技能定义文件

├── references/         # 参考资料目录

│   └── guide.md

└── scripts/            # 脚本目录

    └── helper.py

单文件布局：

data/skills/<skill-id>.toml

skill.toml 文件使用 TOML 格式定义技能的元数据、工具列表、激活条件等属性。以下是一个完整的技能定义示例：

id = "outfit-output-skill"

name = "Outfit Output Skill"

description = "负责最终视觉输出，包括生成穿搭、试穿和发送最终结果。"

activation = "on_demand"

compatibility = "outfit-agent>=1.0"

license = "internal"



tools = [

    "generate_outfit",

    "generate_image",

    "generate_tryon_tool",

    "send_outfits",

    "send_image_message",

]



allowed_tools = [

    "generate_outfit",

    "generate_image",

    "generate_tryon_tool",

    "send_outfits",

    "send_image_message",

]



[channels.wechat_kefu]

tools = []



[channels.wechat_miniapp]

tools = ["edit_outfit"]



[metadata]

category = "output"

owner = "main-agent"



[instructions]

zh = "skills/outfit_output_skill.zh.txt"

en = "skills/outfit_output_skill.en.txt"

Sources: data/skills/example/skill.toml, data/skills/outfit-output-skill/skill.toml

技能定义字段说明

Sources: app/skills/loader.py, data/skills/buyer-wardrobe-skill/skill.toml

技能加载流程

技能加载由 SkillRegistry 类负责，它在应用启动时扫描 data/skills/ 目录，解析所有技能定义文件并构建 RegisteredSkill 对象。

flowchart TD

    A[应用启动] --> B[init_skill_registry]

    B --> C[SkillRegistry.load_all]

    C --> D[扫描 data/skills/ 目录]

    D --> E{发现技能定义文件}

    E -->|*.toml| F[解析单文件技能]

    E -->|*/skill.toml| G[解析目录式技能]

    F --> H[_build_skill]

    G --> H

    H --> I{检查 enable 字段}

    I -->|false| J[跳过]

    I -->|true| K[读取 TOML 配置]

    K --> L[解析工具列表]

    L --> M[解析渠道工具配置]

    M --> N[解析提示词配置]

    N --> O[发现 references/scripts 资源]

    O --> P[从 TOOL_REGISTRY 解析工具对象]

    P --> Q[创建 RegisteredSkill 对象]

    Q --> R[存入 _skills 字典]

SkillRegistry 支持两种加载模式：

1. 1. 基础加载（user_id=None）：加载默认配置，结果缓存在 _skills 字典
2. 2. 灰度加载（user_id 指定）：根据用户 ID 应用灰度覆盖，结果缓存在 _variant_skills 字典

Sources: app/skills/loader.py, app/skills/loader.py

RegisteredSkill 数据结构

RegisteredSkill 是技能加载后的核心数据结构，使用 @dataclass(frozen=True) 定义为不可变对象：

@dataclass(frozen=True)

class RegisteredSkill:

    id: str                              # 技能 ID

    name: str                            # 技能名称

    description: str                     # 技能描述

    activation: str                      # 激活模式: "on_demand" | "always"

    definition_path: Path                # 定义文件路径

    source_path: Path                    # 技能源目录路径

    enabled: bool                        # 是否启用

    tool_names: list[str]                # 全渠道工具名列表

    channel_tools: dict[str, list[str]]  # 渠道专属工具映射

    tools: list[Any]                     # 已解析的工具对象列表

    references: list[str]                # 参考资料文件名列表

    scripts: list[str]                   # 脚本文件名列表

    allowed_tools: list[str] | None      # 声明的工具边界

    metadata: dict[str, Any] | None      # 元数据

    compatibility: str | None            # 兼容性声明

    license: str | None                  # 许可证

    instructions_prompt: str | None      # 提示词文件路径

    skill: Skill                         # agno Skill 实例

    raw_cfg: dict[str, Any]              # 原始 TOML 配置

RegisteredSkill 提供了以下关键方法用于工具查询：

• - all_tool_names()：返回所有工具名（全渠道 + 各渠道专属工具的并集）
• - tools_for_channel(channel)：返回指定渠道可用的工具名列表
• - allowed_channels_for_tool(tool_name)：返回某工具被限制使用的渠道列表
• - is_tool_enabled_for_channel(tool_name, channel)：判断工具在指定渠道是否可用

Sources: app/skills/loader.py

技能包（SkillBundle）构建

当主 Agent 需要使用技能时，通过 SkillRegistry.build_bundle() 方法将多个技能打包为 SkillBundle 对象。SkillBundle 聚合了所有选中技能的工具、提示词和访问工具。

flowchart LR

    A[team_cfg.skills 列表] --> B[SkillRegistry.build_bundle]

    C[SkillRegistry._skills] --> B

    B --> D[遍历 skill_names]

    D --> E[查找 RegisteredSkill]

    E --> F[收集工具对象]

    E --> G[收集 always 技能的提示词]

    E --> H[收集 always 技能的内联指令]

    F --> I[合并 access_tools]

    I --> J[生成 on-demand 技能提示片段]

    J --> K[返回 SkillBundle]

SkillBundle 包含以下关键属性：

team_tools 属性返回 tools 和 access_tools 的合并列表，用于注入主 Agent。

Sources: app/skills/loader.py

技能激活模式

技能支持两种激活模式，通过 activation 字段配置：

on_demand（按需激活）

按需技能不会自动注入系统提示词，而是通过生成提示片段告知 Agent 有哪些可用技能。Agent 在运行时通过调用 get_skill_instructions(skill_name="...") 工具动态加载技能指令。

提示片段格式如下：

<skills_system>

Available on-demand skills are listed below.

Before using a skill's worldview or business tools, call get_skill_instructions(skill_name="...").

Loading a skill activates it in dialog state for the current workflow step.

- nuva-soul: Nuva 的前台人格与护短表达层。

- style-advisor-skill: 负责时尚穿搭方案推荐、评论、审美解释和方案比较。

- buyer-wardrobe-skill: 负责搜单品、找同款、购物链接、衣橱管理。

- outfit-output-skill: 负责最终视觉输出，包括生成穿搭、试穿和发送最终结果。

</skills_system>

当前系统中的四个业务技能均采用 on_demand 模式：

Sources: app/skills/loader.py, data/team/main.toml

always（常驻激活）

常驻技能的指令提示词会自动注入主 Agent 的系统提示词中，无需 Agent 主动调用。always_instruction_prompts 列表中的提示词路径会被传递给 build_main_team_instructions()，在每次构建指令时自动加载。

Sources: app/skills/loader.py, app/agents/main_team_prompt.py

技能运行时访问工具

技能系统提供了三个运行时访问工具，使 Agent 能够在对话过程中动态加载技能内容：

# app/skills/access_tools.py



async def get_skill_instructions(run_context, skill_name: str) -> str

async def get_skill_reference(run_context, skill_name: str, reference_name: str) -> str

async def get_skill_script(run_context, skill_name: str, script_name: str) -> str

当 Agent 调用 get_skill_instructions 时，系统会：

1. 1. 从 SkillRegistry 查找指定技能
2. 2. 调用 activate_skill(session_state, skill.id) 将技能写入对话状态
3. 3. 加载并返回技能的指令提示词

activate_skill() 函数会更新对话状态中的 active_skill 和 loaded_skills 字段，这直接影响工具门控机制。

Sources: app/skills/access_tools.py, app/dialog_state.py

工具门控机制

技能系统与工具门控机制紧密配合，确保技能拥有的工具只有在技能被激活后才能被调用。这是防止 Agent 在错误的业务阶段调用不相关工具的关键安全机制。

sequenceDiagram

    participant Agent as 主 Agent

    participant Factory as 工具工厂

    participant DialogState as 对话状态

    participant SkillRegistry as 技能注册表



    Agent->>Factory: 调用工具 (如 search_items)

    Factory->>SkillRegistry: get_tool_owning_skills("search_items")

    SkillRegistry-->>Factory: ["buyer-wardrobe-skill"]

    Factory->>DialogState: get_dialog_state(session_state)

    DialogState-->>Factory: {active_skill: "style-advisor-skill"}

    Factory->>Factory: is_tool_allowed() 检查

    Note over Factory: active_skill ≠ owning_skill

    Factory-->>Agent: 返回 tool_not_available 错误

工具门控的判断逻辑在 is_tool_allowed() 函数中实现：

def is_tool_allowed(session_state, *, tool_name, owning_skills) -> bool:

    if tool_name in ALWAYS_ALLOWED_TOOLS:  # 常驻白名单工具

        return True

    owners = [item for item in (owning_skills or []) if item]

    if not owners:  # 无技能拥有此工具

        return True

    active_skill = get_dialog_state(session_state).get("active_skill")

    return bool(active_skill) and active_skill in owners

常驻白名单工具（不受门控限制）：

• - run_outfit_flow、update_user_profile、new_session
• - send_text_message、send_miniprogram_message
• - get_skill_instructions、get_skill_reference、get_skill_script
• - get_dialog_state、update_dialog_state

Sources: app/dialog_state.py, app/dialog_state.py, app/tools/factory.py

渠道工具适配

技能支持为不同渠道配置专属工具，通过 skill.toml 中的 [channels.<channel_name>] 段落定义。运行时，tools_for_channel() 方法会合并全渠道工具和指定渠道的专属工具。

以 outfit-output-skill 为例：

• - 全渠道工具：generate_outfit, generate_image, generate_tryon_tool, send_outfits, send_image_message
• - 微信客服渠道：无额外工具（tools = []）
• - 小程序渠道：额外增加 edit_outfit

渠道工具适配在两个层面生效：

1. 1. 技能层面：RegisteredSkill.is_tool_enabled_for_channel() 检查工具是否在当前渠道可用
2. 2. 主 Agent 层面：MainTeamConfig.is_tool_enabled_for_channel() 检查主 Agent 级别的渠道限制

Sources: data/skills/outfit-output-skill/skill.toml, app/skills/loader.py

灰度发布支持

技能配置支持按用户 ID 进行灰度覆盖。SkillRegistry 使用灰度配置服务（GrayConfigService）对技能定义文件应用 TOML 级别的覆盖。

灰度流程：

1. 1. 计算灰度签名：遍历所有技能定义路径，构建 skill:<path> 格式的 scope 列表
2. 2. 查询灰度规则：根据用户 ID 和 scope 列表查询匹配的灰度规则
3. 3. 合并配置：使用 get_toml_overlay() 将灰度规则中的 TOML 片段合并到基础配置
4. 4. 缓存变体：灰度后的技能映射按签名缓存在 _variant_skills 字典

# 灰度覆盖示例：为用户 42 增加工具

# GrayRuleRecord:

#   namespace="skill"

#   path="skills/buyer-wardrobe-skill/skill.toml"

#   users_expr="42"

#   content='tools = ["search_items", "send_items"]'

灰度覆盖会同时影响 build_bundle() 和 get_tool_owning_skills() 的结果，确保灰度用户的工具门控逻辑与配置一致。

Sources: app/skills/loader.py, tests/test_skill_support.py

主 Agent 集成

技能系统在主 Agent 构建时（_build_agent()）完成集成。集成过程如下：

flowchart TD

    A[_build_agent] --> B[加载 team_cfg]

    B --> C[获取技能注册表]

    C --> D[build_bundle: 打包技能]

    D --> E[获取主 Agent 基础工具]

    E --> F[合并工具列表]

    F --> G[构建运行时工具解析器]

    G --> H[构建指令解析器]

    H --> I[注入提示片段到 additional_context]

    I --> J[创建 Agent 实例]

关键集成点：

1. 1. 工具合并：主 Agent 的基础工具（来自 team_cfg.tools）与技能工具（来自 SkillBundle.team_tools）合并
2. 2. 指令注入：always 技能的提示词自动注入系统指令；on_demand 技能的提示片段注入 additional_context
3. 3. 运行时工具解析：通过 _resolve_runtime_tools 函数，每次运行时根据用户 ID 动态解析工具描述

Sources: app/agent.py, app/agent.py

热重载机制

技能系统支持运行时热重载，通过 reload_agents() 函数触发：

def reload_agents() -> None:

    init_tool_registry()      # 重建工具注册表

    init_skill_registry()     # 重建技能注册表

    get_main_team_loader().reload()  # 重载团队配置

    _agent_variants.clear()   # 清除 Agent 缓存

    _agent = _build_agent(leader_runtime=leader_runtime)  # 重建 Agent

热重载会清除所有缓存（包括灰度变体缓存），重新扫描 data/skills/ 目录并重建整个 Agent。这使得在不重启服务的情况下，修改技能定义文件或提示词文件即可生效。

Sources: app/agent.py, app/skills/loader.py

技能提示词管理

技能的指令提示词存放在 data/prompts/skills/ 目录下，通过 skill.toml 中的 [instructions] 段落引用。提示词文件支持中英文双语，运行时根据用户语言偏好动态选择。

当前系统中的技能提示词文件：

提示词内容定义了技能的职责边界、执行规则、输出格式等。Agent 在加载技能指令后，必须严格按照提示词中的约束执行。

Sources: app/skills/loader.py, data/prompts/skills/

技能资源文件

技能可以携带参考资料（references）和脚本（scripts）两类资源文件，分别存放在技能源目录的 references/ 和 scripts/ 子目录中。资源文件在技能加载时自动发现，并通过运行时访问工具提供给 Agent。

• - 参考资料：通过 get_skill_reference(skill_name, reference_name) 读取，适用于分类法、指南等静态知识
• - 脚本文件：通过 get_skill_script(skill_name, script_name) 读取，适用于辅助计算或数据处理逻辑

Sources: app/skills/loader.py, app/skills/access_tools.py

下一步

• - 了解技能工具门控的详细实现机制，请参阅 技能工具门控机制
• - 了解工具注册与工厂模式，请参阅 工具注册与工厂
• - 了解对话状态如何影响技能激活，请参阅 Dialog State 设计
• - 了解如何新增技能模块，请参阅 新增技能模块

<a id="8-ji-neng-gong-ju-men-kong-ji-zhi"></a>

---

技能工具门控机制

• - Section: 架构与核心概念
• - Group: Agent 与技能系统
• - Source File: 8-ji-neng-gong-ju-men-kong-ji-zhi.md

本文档详细阐述 Outfit Agent 中技能（Skill）与工具（Tool）之间的访问控制机制——即"门控"（Gating）。门控机制确保工具只在正确的技能上下文和正确的渠道中可被调用，防止 Agent 跨技能边界误用工具，同时支持按渠道精细控制工具可见性。

门控架构总览

Outfit Agent 采用单主 Agent 架构，所有技能的工具在运行时被注入同一个 Agent。门控机制在工具调用链路中提供两层运行时校验：技能归属门控和渠道可用性门控。当 Agent 调用任意业务工具时，工具包装层（_wrap_tool_entrypoint）会依次执行这两层检查，只有全部通过后工具才能真正执行。

flowchart TD

    A["Agent 调用工具"] --> B["工具包装层\n(_wrap_tool_entrypoint)"]

    B --> C{"渠道门控检查\n(_tool_available_for_current_channel)"}

    C -->|"团队级渠道限制不通过"| D["返回 tool_not_available_on_channel 错误"]

    C -->|"技能级渠道限制不通过"| D

    C -->|"通过"| E{"技能归属门控检查\n(is_tool_allowed)"}

    E -->|"工具在 ALWAYS_ALLOWED_TOOLS 中"| F["工具正常执行"]

    E -->|"无技能归属（公共工具）"| F

    E -->|"当前 active_skill 匹配工具归属技能"| F

    E -->|"不匹配"| G["返回 tool_not_available 错误"]

    F --> H["执行工具入口函数"]



    style D fill:#f8d7da,stroke:#dc3545,color:#721c24

    style G fill:#f8d7da,stroke:#dc3545,color:#721c24

    style F fill:#d4edda,stroke:#28a745,color:#155724

    style H fill:#d4edda,stroke:#28a745,color:#155724

Sources: factory.py, dialog_state.py

技能归属门控

技能归属门控是门控机制的第一层（在渠道门控之后执行）。其核心思想是：工具必须归属于当前处于激活状态的技能，才能被调用。

归属关系建立

每个技能在 skill.toml 中通过 tools 字段声明它拥有的工具列表。SkillRegistry 加载时会扫描所有技能配置，并建立工具名到技能 ID 的反向映射。

# data/skills/buyer-wardrobe-skill/skill.toml

id = "buyer-wardrobe-skill"

tools = [

    "search_items",

    "search_items_batch",

    "search_wardrobe",

    "get_item_detail",

    "save_item_batch",

    # ... 更多工具

]

get_tool_owning_skills 方法遍历所有已注册技能，返回包含指定工具的技能 ID 列表。例如查询 "search_items" 的归属时，返回 ["buyer-wardrobe-skill"]。

Sources: loader.py, buyer-wardrobe-skill/skill.toml

门控判定逻辑

is_tool_allowed 函数实现了三层短路判定：

ALWAYS_ALLOWED_TOOLS 是一个固定的白名单集合，包含以下工具：run_outfit_flow、update_user_profile、new_session、send_text_message、send_miniprogram_message、get_skill_instructions、get_skill_reference、get_skill_script、get_dialog_state、update_dialog_state。这些工具是系统运转的基础设施，不应被门控阻断。

Sources: dialog_state.py, dialog_state.py

技能激活流程

技能处于被动激活状态——Agent 需要主动调用访问工具（Access Tools）来激活技能。access_tools.py 提供了三个访问工具：

1. 1. get_skill_instructions(skill_name)：读取技能的提示词指令，并将该技能标记为当前会话的 active_skill
2. 2. get_skill_reference(skill_name, reference_name)：读取技能附属的参考文档，同时激活技能
3. 3. get_skill_script(skill_name, script_name)：读取技能附属的脚本文件，同时激活技能

调用任一访问工具都会触发 activate_skill，将技能 ID 写入 dialog_state.active_skill，同时追加到 dialog_state.loaded_skills 列表。这意味着同一时刻只有一个技能处于激活状态——后激活的技能会替换先前的激活状态。

Agent 在对话过程中会收到一段由 _build_prompt_snippet 生成的系统提示，其中列出所有可用的 on_demand 技能，并引导 Agent 在使用技能前先调用 get_skill_instructions。

Sources: access_tools.py, dialog_state.py, loader.py

错误响应格式

当技能归属门控拒绝工具调用时，返回标准化的 JSON 错误：

{

  "error": "tool_not_available",

  "tool_name": "search_items",

  "required_skill": ["buyer-wardrobe-skill"],

  "message": "Tool 'search_items' requires activating one of: buyer-wardrobe-skill"

}

Agent 收到此错误后，应先调用 get_skill_instructions(skill_name="buyer-wardrobe-skill") 激活对应技能，再重新调用该工具。

Sources: dialog_state.py

渠道可用性门控

渠道门控是门控机制的第二层（在技能归属门控之前执行），负责确保工具只在指定的渠道中可用。渠道门控有两个层级：团队级和技能级。

团队级渠道限制

在 data/team/main.toml 中，通过 [channels.<channel_name>] 段为不同渠道声明专属工具。运行时，MainTeamConfig 的 allowed_channels_for_tool 方法会返回工具被允许使用的渠道列表。

# data/team/main.toml

tools = ["run_outfit_flow", "update_user_profile", "new_session", ...]



[channels.wechat_kefu]

tools = ["send_miniprogram_message"]

如果 send_miniprogram_message 被声明在 [channels.wechat_kefu] 下，则该工具只能在 wechat_kefu 渠道中使用，在 wechat_miniapp 渠道中调用将返回错误。

Sources: main.toml, team_loader.py

技能级渠道限制

每个技能也可以在 skill.toml 中通过 [channels.<channel_name>] 为特定渠道声明额外的或受限的工具。例如 outfit-output-skill 配置：

# data/skills/outfit-output-skill/skill.toml

tools = ["generate_outfit", "generate_image", "generate_tryon_tool", "send_outfits", "send_image_message"]



[channels.wechat_kefu]

tools = []



[channels.wechat_miniapp]

tools = ["edit_outfit"]

此配置表明 edit_outfit 是该技能的渠道专属工具，仅在 wechat_miniapp 渠道中可用。当 active_skill 为 outfit-output-skill 且当前渠道为 wechat_kefu 时，调用 edit_outfit 会被技能级渠道门控拦截。

RegisteredSkill 通过以下方法管理渠道工具：

• - all_tool_names()：返回基础工具 + 所有渠道工具的合并列表
• - tools_for_channel(channel)：返回指定渠道下该技能可用的工具名列表
• - allowed_channels_for_tool(tool_name)：返回允许使用该工具的渠道列表（空列表表示全渠道可用）
• - is_tool_enabled_for_channel(tool_name, channel)：判定工具在指定渠道是否可用

Sources: outfit-output-skill/skill.toml, loader.py

渠道门控执行流程

_tool_available_for_current_channel 函数按以下顺序检查：

1. 1. 从当前会话的 session_state 中解析出 current_channel
2. 2. 团队级检查：查询团队配置的 allowed_channels_for_tool，若工具被限制且当前渠道不在允许列表中，直接返回错误
3. 3. 技能级检查：如果存在 active_skill，查询该技能的 allowed_channels_for_tool，若工具被限制且当前渠道不在允许列表中，返回错误
4. 4. 两级检查均通过，返回 None（允许通过）

渠道门控错误的 JSON 格式：

{

  "error": "tool_not_available_on_channel",

  "tool_name": "edit_outfit",

  "allowed_channels": ["wechat_miniapp"],

  "message": "Tool 'edit_outfit' is only available for channels: wechat_miniapp"

}

Sources: factory.py, factory.py

工具包装与注入

门控机制通过 build_tool 和 _wrap_tool_entrypoint 统一注入到每个工具中。这是一个透明的包装层，对工具实现代码完全无侵入。

build_tool 包装流程

build_tool 函数接收一个原始的工具入口函数，执行以下步骤：

1. 1. 调用 _wrap_tool_entrypoint 包装入口函数，注入渠道门控和技能归属门控逻辑
2. 2. 通过 Function.from_callable 创建 agno Function 对象
3. 3. 从 tools/descriptions.toml 加载工具描述
4. 4. 处理 __tool_stop_after_tool_call__、__tool_stop_after_tool_call_if__、__tool_show_result__ 等装饰属性

包装后的工具在每次调用时会：

1. 1. 绑定 Agno 上下文（用于链路追踪）
2. 2. 执行渠道门控检查
3. 3. 执行技能归属门控检查
4. 4. 两层检查均通过后，调用原始入口函数
5. 5. 记录追踪信息

Sources: factory.py, factory.py

工具注入时机

在 _build_agent 函数中，工具通过两个来源合并注入到主 Agent：

1. 1. 团队工具：get_main_agent_tools(team_cfg) 根据团队配置解析基础工具列表
2. 2. 技能工具：team_skill_bundle.team_tools 包含所有技能声明的工具和访问工具

合并后，工具列表通过 materialize_tools_for_user 为当前用户加载个性化的工具描述（支持灰度覆盖），最终注入 Agent。

Sources: agent.py

完整门控示例

以下通过一个端到端示例展示门控机制的完整运作：

sequenceDiagram

    participant U as 用户

    participant A as Agent

    participant GW as 工具包装层

    participant DS as DialogState



    U->>A: "帮我搜一下白色连衣裙"

    Note over A: Agent 决策：需要使用 search_items 工具

    A->>GW: 调用 search_items(...)

    GW->>DS: 查询 active_skill

    DS-->>GW: active_skill = null（未激活任何技能）

    GW-->>A: 返回 tool_not_available 错误

    Note over A: Agent 意识到需要先激活技能



    A->>GW: 调用 get_skill_instructions("buyer-wardrobe-skill")

    GW->>DS: 检查 get_skill_instructions 是否在 ALWAYS_ALLOWED_TOOLS

    Note over GW: ✅ 在白名单中，允许执行

    GW->>DS: activate_skill("buyer-wardrobe-skill")

    DS-->>GW: active_skill = "buyer-wardrobe-skill"

    GW-->>A: 返回技能指令文本



    A->>GW: 调用 search_items(query="白色连衣裙")

    GW->>DS: 查询 active_skill

    DS-->>GW: active_skill = "buyer-wardrobe-skill"

    GW->>GW: search_items 归属于 buyer-wardrobe-skill ✅

    GW-->>A: 执行成功，返回搜索结果

    A-->>U: "找到以下白色连衣裙..."

Sources: test_skill_tool_gating.py

灰度配置支持

门控机制完整支持灰度配置覆盖。SkillRegistry 和 MainTeamLoader 都通过 GrayConfigService 为特定用户加载覆盖版本的技能配置。

灰度路径为 skills/<skill-id>/skill.toml，覆盖后可以改变技能的 tools 列表、channels 配置等。这意味着运营人员可以针对特定用户开启或关闭某些工具、调整渠道限制，而无需修改代码或重启服务。

SkillRegistry 通过 _gray_signature 方法为每个用户构建唯一的签名，缓存该用户对应的技能变体，避免重复加载。invalidate 方法用于在灰度配置变更时清除缓存。

Sources: loader.py, loader.py, test_skill_support.py

关键设计决策

Sources: dialog_state.py, factory.py, loader.py

参考与导航

• - 前置阅读：技能加载与管理 了解技能的定义、目录结构与加载流程
• - 前置阅读：单主 Agent 架构 了解技能工具注入的上下文
• - 后续阅读：工具注册与工厂 了解工具的注册机制与 build_tool 详情
• - 后续阅读：工具描述管理 了解工具描述的加载与灰度覆盖
• - 后续阅读：Dialog State 设计 了解 active_skill 等状态字段的完整设计

<a id="9-dialog-state-she-ji"></a>

---

Dialog State 设计

• - Section: 架构与核心概念
• - Group: 对话状态与 SOP 流程
• - Source File: 9-dialog-state-she-ji.md

Dialog State 是 Outfit Agent 的会话级有限状态机，它将一次对话交互的上下文——从技能激活、信息收集、SOP 阶段推进到穿搭工作流生命周期——统一编码为一个嵌套字典，存储于 Agno Session 的 session_state["dialog_state"] 键下。该设计的核心目标是：让 Agent 在每一轮对话中都能"知道自己在哪里"，同时为工具门控、提示词注入和后台可观测性提供统一的状态基底。

整体架构与数据流

Dialog State 在系统中的位置可以用以下数据流来概括：Agent 工具调用（读/写）→ session_state["dialog_state"]（内存态）→ Agno Session 持久化 → 后台异步写入（persist_dialog_state_update）。读取路径覆盖提示词注入、工具门控和 Trace 属性构建三条并行分支。

flowchart LR

    subgraph Agent Runtime

        A[Agent LLM] -->|工具调用| B[dialog_state_tools]

        B -->|读写| C[session_state<br/>dialog_state]

    end



    subgraph Prompt Injection

        C -->|describe_dialog_state_for_prompt| D[profile_hook<br/>additional_context]

    end



    subgraph Tool Gating

        C -->|is_tool_allowed| E[factory.py<br/>build_tool wrapper]

    end



    subgraph Observability

        C -->|_dialog_state_summary| F[live_runs.py]

        C -->|_build_trace_attributes| G[Trace Attributes]

    end



    subgraph Async Persistence

        H[followup_intent<br/>workflow_callback] -->|persist_dialog_state_update| I[dialog_state_store]

        I -->|asave_session| J[(Agno Session DB)]

    end

Sources: app/dialog_state.py, app/services/dialog_state_store.py, app/tools/dialog_state_tools.py

状态字典结构

DEFAULT_DIALOG_STATE 定义了 Dialog State 的完整骨架。每个字段承担明确的职责，不存在语义重叠。以下是字段分类与类型说明：

Sources: app/dialog_state.py

核心操作函数

状态读取与初始化

系统提供两个读取函数，分别服务于可写场景和只读场景，选择不同的合并策略：

• - ensure_dialog_state(session_state) — 可写路径。在现有状态与 DEFAULT_DIALOG_STATE 之间做浅合并（现有字段优先），然后直接写回 session_state["dialog_state"]。适合工具执行过程中需要修改状态的场景。
• - get_dialog_state(session_state) — 只读路径。同样做合并但不写回原始 session_state，返回一个独立的字典副本。适合提示词注入、Trace 属性构建等不应产生副作用的场景。

两个函数都会调用 _normalize_skill_list() 对 loaded_skills 做去重保序处理，并确保 collected_slots 始终为 dict 类型。

Sources: app/dialog_state.py

状态更新

update_dialog_state_fields() 是状态写入的唯一核心函数，所有更新路径最终都汇聚于此。它接受丰富的 keyword-only 参数，支持增量更新和条件清除两种模式：

• - 增量更新：传入非 None 的参数直接覆盖对应字段。collected_slots 采用浅合并策略（新键覆盖旧键，旧键保留）。
• - 条件清除：通过 clear_* 布尔参数显式清空对应字段（如 clear_outfit_run_id=True 将 outfit_run_id 设为 None）。
• - 全局重置：clear_other=True 会将整个状态回退到 DEFAULT_DIALOG_STATE，仅保留本次调用明确写入的新值。适用于用户切换到全新任务的场景。

值得注意的是，该函数内嵌了一条状态归一化逻辑：当 sop_stage 为 await_outfit_confirmation 且 pending_confirmation.action 为 run_outfit_flow 时，如果 outfit_status 为空或仍为 running，系统会自动将其归一化为 awaiting_confirmation，同时清空 outfit_run_id 和 outfit_result_summary。这条规则确保了"等待确认"状态的一致性。

Sources: app/dialog_state.py

技能激活

activate_skill() 是技能加载的专用入口，执行以下原子操作：

1. 1. 调用 ensure_dialog_state() 确保状态已初始化
2. 2. 将 skill_name 追加到 loaded_skills（若不存在）
3. 3. 将 active_skill 设为 skill_name
4. 4. 可选地同时写入 sop_stage 和 active_scene

Sources: app/dialog_state.py

工具门控机制

Dialog State 驱动了系统的技能级工具门控，这是防止 Agent 在错误的 SOP 阶段调用不相关工具的核心安全机制。

门控规则

is_tool_allowed(session_state, tool_name, owning_skills) 的判定逻辑：

1. 1. 若 tool_name 在 ALWAYS_ALLOWED_TOOLS 集合中 → 始终放行
2. 2. 若 owning_skills 为空（工具不属于任何技能）→ 始终放行
3. 3. 否则，检查 active_skill 是否在 owning_skills 列表中 → 匹配则放行，否则拒绝

ALWAYS_ALLOWED_TOOLS 包含 10 个系统级工具：run_outfit_flow、update_user_profile、new_session、send_text_message、send_miniprogram_message、get_skill_instructions、get_skill_reference、get_skill_script、get_dialog_state、update_dialog_state。这些工具在任何 SOP 阶段都可被调用。

flowchart TD

    A[Agent 调用工具] --> B{tool_name 在<br/>ALWAYS_ALLOWED_TOOLS?}

    B -->|是| C[✅ 放行]

    B -->|否| D{owning_skills 为空?}

    D -->|是| C

    D -->|否| E{active_skill 在<br/>owning_skills 中?}

    E -->|是| C

    E -->|否| F[❌ 拒绝<br/>返回 tool_not_available 错误]

工厂层封装

app/tools/factory.py 中的 build_tool() 将 is_tool_allowed() 的检查逻辑封装在每个工具的入口 wrapper 中。每次工具调用都会经历以下检查链：渠道门控 → 技能门控 → 实际执行。门控失败时返回结构化 JSON 错误，包含 error、tool_name、required_skill 和 message 字段，供 Agent 理解拒绝原因并调整策略。

Sources: app/dialog_state.py, app/tools/factory.py

SOP 阶段与工作流生命周期

sop_stage 字段是 SOP 流水线的阶段标识符，它驱动了从需求探索到视觉交付的完整主线漏斗。outfit_status 字段则独立追踪后台工作流的运行生命周期。两者协同工作但语义独立：

sop_stage 阶段表

outfit_status 状态表

状态转换流程

stateDiagram-v2

    [*] --> collecting_idea: activate_skill / 信息收集

    collecting_idea --> await_idea_confirmation: 构想确认

    await_idea_confirmation --> searching_items: 搜品开始

    searching_items --> await_search_confirmation: 搜品完成

    await_search_confirmation --> await_outfit_confirmation: 任务单写入



    await_outfit_confirmation --> outfit_running: run_outfit_flow

    outfit_running --> s3_visual_delivered: workflow 完成

    outfit_running --> outfit_failed: workflow 失败

    outfit_running --> await_outfit_confirmation: 用户中断



    s3_visual_delivered --> [*]

    outfit_failed --> await_outfit_confirmation: 重新确认

    outfit_failed --> collecting_idea: 回退到 S1

Sources: app/dialog_state.py, app/services/outfit_workflow_service.py, app/tools/outfit_workflow_tools.py, data/prompts/agents/main_team.zh.txt

待确认机制 (Pending Confirmation)

pending_confirmation 字典是 Dialog State 中的用户确认门控，用于在关键 SOP 阶段断点处强制等待用户的明确意图。其结构包含两个固定键：

• - action：标识待确认的操作类型（如 run_outfit_flow、search_items、confirm_search_results）
• - summary：面向 Agent 的确认提示文本

该机制在以下场景中被激活：

1. 1. 穿搭任务发车确认：当 outfit_task 写入后，pending_confirmation.action 设为 run_outfit_flow，sop_stage 推进到 await_outfit_confirmation。只有当用户明确表达"开始"等意图时，Agent 才被允许执行 run_outfit_flow。
2. 2. 搜品结果确认：buyer-wardrobe-skill 完成搜品后，通过 pending_confirmation 等待用户对结果的认可。
3. 3. 工作流中断恢复：当用户在工作流运行期间发送新消息导致中断时，系统通过 persist_dialog_state_update() 将状态回退到 await_outfit_confirmation，并写入中断相关的 pending_confirmation。

Sources: app/dialog_state.py, app/tools/outfit_workflow_tools.py, app/wechat_followup_intent.py

提示词注入

Dialog State 不仅是后台状态管理的工具，还通过 describe_dialog_state_for_prompt() 函数直接注入到 Agent 的系统提示词中。该函数被 app/agents/profile_hook.py 调用，在每次 Agent 运行前将有意义的状态字段序列化为 JSON 文本，拼接到 agent.additional_context 中。

注入的字段包括：active_skill、loaded_skills、active_scene、stage、sop_stage、collection_plan、collected_slots、idea_summary、outfit_task、outfit_status、outfit_run_id、outfit_result_summary、pending_confirmation。若所有字段均为空/None，该函数返回 None 以避免污染提示词。

这种设计使得 Agent 在每一轮对话开始时就完全了解当前的 SOP 进度和待处理事项，无需额外的工具调用来"回忆"状态。

Sources: app/dialog_state.py, app/agents/profile_hook.py

异步持久化

当 Dialog State 需要在 Agent 运行上下文之外被修改时（如微信 follow-up intent 仲裁、workflow 回调），app/services/dialog_state_store.py 提供了 persist_dialog_state_update() 函数。该函数的执行路径为：

1. 1. 通过 aread_or_create_session() 从 Agno Session Store 加载会话
2. 2. 通过 load_session_state() 恢复 session_state
3. 3. 调用 update_dialog_state_fields() 应用增量更新
4. 4. 将更新后的状态写回 session_data["session_state"]
5. 5. 通过 asave_session() 持久化

该函数接受 session_id、user_id 和 update_kwargs（直接透传给 update_dialog_state_fields()），确保了 Agent 运行时和异步回调对 Dialog State 的操作使用同一套合并逻辑。

Sources: app/services/dialog_state_store.py

可观测性

Dialog State 的关键字段被系统性地暴露到两个可观测性通道：

1. 1. 实时运行快照（app/live_runs.py）：_dialog_state_summary() 提取 11 个核心字段的摘要，用于 Dashboard 实时展示会话状态。
2. 2. Trace 属性（app/tools/factory.py）：_build_trace_attributes() 将 active_skill、active_scene、sop_stage、outfit_status、outfit_run_id、pending_confirmation 和 loaded_skills 作为 Span 属性附加到每次工具调用的 Trace 中，支持链路追踪中的状态关联查询。

Admin 端的 Trace 查询接口（_build_session_summary）也会从 session_data 中提取 Dialog State 字段，为管理面板提供完整的会话上下文视图。

Sources: app/live_runs.py, app/tools/factory.py, tests/test_admin_traces_dialog_state.py

Agent 工具接口

Agent 通过两个工具函数与 Dialog State 交互，这两个函数注册在 STATE_TOOLS 分组中，并列入 ALWAYS_ALLOWED_TOOLS：

update_dialog_state 工具函数是 update_dialog_state_fields() 的薄封装，负责将 LLM 传入的 JSON 字符串参数解析为字典，对 outfit_task 执行 normalize_outfit_task_payload_data() 标准化，以及将 pending_confirmation_action + pending_confirmation_summary 组装为 pending_confirmation 字典。

工具描述定义在 data/tools/descriptions.toml 中，支持按用户维度的灰度覆盖。

<a id="10-sop-jie-duan-tui-jin"></a>

---

SOP 阶段推进

• - Section: 架构与核心概念
• - Group: 对话状态与 SOP 流程
• - Source File: 10-sop-jie-duan-tui-jin.md

SOP 阶段推进是 Outfit Agent 的对话流程控制核心，它通过 dialog_state.sop_stage 字段将一次完整的穿搭交互——从需求探索、信息收集、任务确认到后台工作流执行——编码为一条可观测、可回退、可中断的有限状态管线。每个阶段不仅决定了 Agent 的合法战术动作，还驱动了工具门控、用户确认断点和异步工作流生命周期管理。

主线漏斗与 SOP 阶段映射

系统将一次穿搭交互划分为四个主线漏斗阶段（S1–S4），每个阶段在主 Agent 提示词中定义了明确的战术动作，同时映射到 sop_stage 的具体枚举值。以下是完整的阶段对照：

Sources: app/dialog_state.py, data/prompts/agents/main_team.zh.txt, app/services/outfit_workflow_service.py

状态转换流程

SOP 阶段遵循一条主线正向推进路径，同时在工作流中断和失败场景下支持回退。以下状态图展示了完整的转换关系：

stateDiagram-v2

    [*] --> collecting_idea: activate_skill / 信息收集

    collecting_idea --> await_idea_confirmation: 构想摘要产出

    await_idea_confirmation --> searching_items: 用户确认方向

    searching_items --> await_search_confirmation: 搜品完成

    await_search_confirmation --> await_outfit_confirmation: 任务清单写入



    await_outfit_confirmation --> outfit_running: run_outfit_flow

    outfit_running --> s3_visual_delivered: workflow 完成

    outfit_running --> outfit_failed: workflow 失败

    outfit_running --> await_outfit_confirmation: 用户中断 / followup



    s3_visual_delivered --> [*]

    outfit_failed --> await_outfit_confirmation: 重新确认

    outfit_failed --> collecting_idea: 回退到 S1

Sources: app/services/outfit_workflow_service.py, app/tools/outfit_workflow_tools.py, app/wechat_followup_intent.py

SOP 阶段写入路径

sop_stage 的写入通过 update_dialog_state_fields() 统一汇聚，但触发入口分布在三个层面：Agent 工具调用、工作流服务、以及异步事件回调。

Agent 工具调用路径

主 Agent 通过 update_dialog_state 工具直接推进 SOP 阶段。该工具接受 sop_stage 参数，内部调用 update_dialog_state_fields() 完成写入。主提示词中明确规定了 S1→S2 的关键转换动作：当 collection_plan.missing_slots 已清空时，Agent 调用 update_dialog_state(sop_stage="await_outfit_confirmation", pending_confirmation_action="run_outfit_flow") 将控制权交给用户确认断点。

Sources: app/tools/dialog_state_tools.py, data/prompts/agents/main_team.zh.txt

工作流工具路径

run_outfit_flow 是 await_outfit_confirmation → outfit_running 的唯一合法入口。该工具内置了严格的前置检查——只有当 sop_stage == "await_outfit_confirmation" 且 pending_confirmation.action == "run_outfit_flow" 时才放行。执行成功后，通过 resolve_outfit_workflow_sop_stage() 将工作流状态映射为最终 SOP 阶段：

• - completed 或 partial_failed → s3_visual_delivered
• - 其他状态 → outfit_failed

def resolve_outfit_workflow_sop_stage(status: str | None) -> str:

    normalized_status = str(status or "").strip()

    if normalized_status in _DELIVERED_WORKFLOW_STATUSES:

        return "s3_visual_delivered"

    return "outfit_failed"

Sources: app/tools/outfit_workflow_tools.py, app/tools/outfit_workflow_tools.py, app/services/outfit_workflow_service.py

异步事件路径

当用户在 outfit_running 阶段发送新消息导致工作流中断时，系统通过两条路径回退 SOP 阶段：

1. 1. wechat_followup_intent.py：当 followup 意图仲裁器判定为 interrupt 动作且检测到穿搭工作流正在运行时，通过 persist_dialog_state_update() 将 sop_stage 回退到 await_outfit_confirmation，同时清除 outfit_run_id 并写入中断相关的 pending_confirmation。
2. 2. outfit_workflow_service.py：工作流内部捕获 asyncio.CancelledError 时，同样回退到 await_outfit_confirmation。服务启动时的 recover_interrupted_workflows() 方法还会扫描数据库中所有 outfit_status == "running" 的会话，根据工作流实际运行状态决定恢复为 s3_visual_delivered 或 await_outfit_confirmation。

Sources: app/wechat_followup_intent.py, app/services/outfit_workflow_service.py, app/services/outfit_workflow_service.py

用户确认断点机制

SOP 管线中的关键阶段转换都经过 pending_confirmation 字段进行门控。该字段包含 action（待确认的操作类型）和 summary（面向 Agent 的确认提示文本）两个固定键。当 sop_stage 为 await_outfit_confirmation 且 pending_confirmation.action 为 run_outfit_flow 时，只有当用户明确表达"开始"等意图后，Agent 才被允许执行 run_outfit_flow。

系统在 update_dialog_state_fields() 中内嵌了一条状态归一化逻辑：当上述两个条件同时满足时，如果 outfit_status 为空或仍为 running，系统会自动将其归一化为 awaiting_confirmation，同时清空 outfit_run_id 和 outfit_result_summary。这条规则确保了"等待确认"状态的一致性，防止残留的工作流运行态干扰后续判断。

Sources: app/dialog_state.py, app/tools/outfit_workflow_tools.py

SOP 阶段与工具门控的协同

sop_stage 与工具门控体系协同工作，共同约束 Agent 在每个阶段的合法行为。factory.py 在构建工具上下文时将 sop_stage 注入到运行时快照中（键名为 sop.stage），供下游工具和提示词注入使用。技能级工具门控由 active_skill 字段驱动，而 SOP 阶段则通过 describe_dialog_state_for_prompt() 直接注入到 Agent 系统提示词中，使 Agent 在每一轮对话开始时就完全了解当前的 SOP 进度和待处理事项，无需额外的工具调用来"回忆"状态。

flowchart LR

    subgraph 每轮对话开始

        A[Agent 启动] --> B[profile_hook]

        B -->|describe_dialog_state_for_prompt| C[系统提示词注入 sop_stage]

    end



    subgraph 工具调用时

        D[Agent LLM] -->|调用工具| E[factory.py]

        E -->|is_tool_allowed| F{active_skill 门控}

        E -->|sop.stage 注入| G[工具上下文]

    end



    subgraph 关键断点

        H[sop_stage == await_outfit_confirmation] --> I{pending_confirmation 检查}

        I -->|通过| J[run_outfit_flow 放行]

        I -->|未通过| K[返回错误]

    end

Sources: app/tools/factory.py, app/dialog_state.py, app/agents/profile_hook.py

工作流生命周期与 SOP 阶段的对应关系

outfit_status 字段独立追踪后台工作流的运行生命周期，与 sop_stage 协同但语义独立。两者之间的对应关系如下：

工作流本身由 OutfitGenerationWorkflow 定义，包含五个顺序 Step：prepare_context → ideate_outfits → select_items → generate_images（并发）→ dispatch_results。dispatch_results 根据成功/失败数量决定最终状态：全部成功 → completed，部分成功 → partial_failed，全部失败 → failed。只有全部成功时才发送总结文案，否则只发送成功部分和失败提示。

Sources: app/workflows/outfit_generation.py, app/services/outfit_workflow_service.py, data/i18n/workflow.zh.json

工作流中断与恢复策略

系统在三个层面处理工作流中断：

前端 followup 中断：当 outfit_running 阶段用户发送新消息时，followup 意图仲裁器根据消息类型（情绪催促、局部微调、意图颠覆、无关闲聊）决定是否中断。中断时通过 persist_dialog_state_update() 回退状态并取消当前 Agno Run。

工作流内部中断：run_outfit_flow 工具在捕获 asyncio.CancelledError 时，将 sop_stage 回退到 await_outfit_confirmation，并将 outfit_result_summary 标记为 interrupted，同时写入 pending_confirmation 以支持用户后续恢复。

服务重启恢复：OutfitWorkflowService.recover_interrupted_workflows() 在服务启动时扫描数据库中所有 outfit_status == "running" 的会话，查询对应工作流的最新运行状态，然后做出恢复决策：如果工作流已完成则更新为最终状态，如果工作流已失败则回退到 await_outfit_confirmation 并提示用户恢复。

Sources: app/wechat_followup_intent.py, app/services/outfit_workflow_service.py, app/services/outfit_workflow_service.py

防打断策略

主提示词在 S2 阶段定义了四类用户消息的防打断策略，这些策略通过 Agent 的提示词层执行，与 SOP 阶段共同维护工作流的原子性：

只有意图颠覆（底层场景或核心意图发生根本性反转）才会触发真正的管线中断，其他三类都通过话术引导用户耐心等待工作流完成。

<a id="11-yong-hu-que-ren-ji-zhi"></a>

---

用户确认机制

• - Section: 架构与核心概念
• - Group: 对话状态与 SOP 流程
• - Source File: 11-yong-hu-que-ren-ji-zhi.md

用户确认机制是 NUVA Fashion OS 中确保穿搭生成工作流可控执行的核心安全机制。该机制通过对话状态管理（Dialog State）和 SOP 阶段控制，在关键操作执行前强制插入用户确认环节，防止未经用户授权的后台任务执行，同时支持工作流中断后的状态恢复。

机制概述与设计原则

用户确认机制的核心设计遵循"显式授权"原则：任何涉及资源密集型操作（如穿搭图生成、单品搜索、试穿渲染）的任务，都必须在用户明确确认后才能执行。这种设计既避免了系统资源的浪费，也确保了用户对生成结果的预期管理。

该机制通过三个核心状态字段实现：

• - sop_stage：SOP 流程阶段标识，标记当前对话在穿搭工作流中的位置
• - pending_confirmation：待确认事项，包含需要用户确认的操作类型和描述
• - outfit_status：穿搭工作流状态，标记后台任务的执行状态

当系统完成需求收集并准备好执行生成任务时，会进入"等待确认"状态；只有用户明确表达"开始生成"等意图后，系统才会执行实际的生成操作。

Sources: dialog_state.py, outfit_workflow_tools.py

核心数据结构与状态流转

用户确认机制的核心数据结构定义在 dialog_state.py 中，通过 DEFAULT_DIALOG_STATE 初始化：

DEFAULT_DIALOG_STATE: dict[str, Any] = {

    "version": 1,

    "active_skill": None,

    "loaded_skills": [],

    "active_scene": None,

    "stage": None,

    "sop_stage": None,

    "collection_plan": None,

    "collected_slots": {},

    "idea_summary": None,

    "outfit_task": None,

    "outfit_status": None,

    "outfit_run_id": None,

    "outfit_result_summary": None,

    "pending_confirmation": None,

    "updated_at": 0,

}

其中 pending_confirmation 字段是一个字典，包含两个关键属性：

• - action：需要确认的操作类型（如 "run_outfit_flow"）
• - summary：向用户展示的确认描述文本

状态流转遵循严格的时序逻辑：

1. 1. 需求收集阶段：系统通过 collection_plan 收集用户需求，sop_stage 为 collecting_requirements
2. 2. 任务准备阶段：需求收集完成后，系统写入 outfit_task，sop_stage 变为 await_outfit_confirmation
3. 3. 用户确认阶段：pending_confirmation.action 设置为 run_outfit_flow，等待用户确认
4. 4. 执行阶段：用户确认后，sop_stage 变为 outfit_running，outfit_status 变为 running
5. 5. 完成阶段：工作流完成后，sop_stage 变为 s3_visual_delivered

Sources: dialog_state.py, dialog_state.py

工具层实现与门控逻辑

用户确认机制在工具层通过 outfit_workflow_tools.py 和 outfit_tools.py 实现严格的门控逻辑。核心函数 _is_outfit_confirmation_ready 负责检查当前状态是否允许执行穿搭工作流：

def _is_outfit_confirmation_ready(

    dialog_state: dict[str, Any],

    *,

    allowed_actions: set[str],

) -> bool:

    pending_confirmation = dialog_state.get("pending_confirmation") or {}

    return (

        dialog_state.get("sop_stage") == "await_outfit_confirmation"

        and pending_confirmation.get("action") in allowed_actions

    )

run_outfit_flow 工具在执行前会进行双重验证：

1. 1. 状态验证：检查 sop_stage 是否为 await_outfit_confirmation
2. 2. 确认验证：检查 pending_confirmation.action 是否为 run_outfit_flow

如果验证失败，工具会返回错误信息，阻止工作流执行。同时，outfit_tools.py 中的 generate_outfit 工具也会检查是否存在待确认的经典工作流，防止绕过确认机制：

if (

    str(collection_plan.get("task_kind") or "").strip() == "classic_outfit"

    and pending_confirmation.get("action") == "run_outfit_flow"

):

    return _build_classic_workflow_bypass_error()

这种双重门控机制确保了即使用户尝试通过其他工具绕过确认，系统也能正确拦截。

Sources: outfit_workflow_tools.py, outfit_workflow_tools.py, outfit_tools.py

工作流中断与状态恢复

用户确认机制的一个重要特性是支持工作流中断后的状态恢复。当用户在穿搭工作流执行期间发送新消息时，系统会通过 wechat_followup_intent.py 中的中断处理逻辑：

1. 1. 检测中断条件：通过 _should_reset_outfit_confirmation_state 函数检测是否需要中断
2. 2. 更新对话状态：将 sop_stage 重置为 await_outfit_confirmation
3. 3. 设置待确认事项：创建新的 pending_confirmation，提示用户是否恢复中断的工作流
4. 4. 取消当前任务：调用 agent.acancel_run 取消正在执行的 agno run

中断后的状态恢复通过 _build_interrupted_update 函数实现：

def _build_interrupted_update(

    *,

    lang: str,

    interrupted_run_id: str | None,

) -> dict[str, object]:

    return {

        "sop_stage": "await_outfit_confirmation",

        "outfit_status": "awaiting_confirmation",

        "clear_outfit_run_id": True,

        "outfit_result_summary": {

            "status": "interrupted",

            "error": "workflow_interrupted_by_followup",

            "interrupted_run_id": interrupted_run_id,

        },

        "pending_confirmation": {

            "action": "run_outfit_flow",

            "summary": get_text(

                "workflow",

                "followup_interrupt_pending_confirmation",

                lang=lang,

            ),

        },

    }

这种设计确保了即使工作流被中断，用户仍然可以选择恢复执行，而不需要重新开始整个流程。

Sources: wechat_followup_intent.py, outfit_workflow_service.py

工具描述与用户交互

用户确认机制在工具描述中明确告知用户触发条件和预期行为。run_outfit_flow 工具的描述强调：

只有当 dialog_state.sop_stage == await_outfit_confirmation 且 pending_confirmation.action == run_outfit_flow 时才能执行；否则会返回错误。

update_dialog_state 工具的描述也明确了如何设置待确认事项：

若需要用户确认，可传入 pending_confirmation_action 与 pending_confirmation_summary；若要清空待确认事项，可传入 clear_pending_confirmation=true。

在用户交互层面，系统会通过提示词指导 Agent 正确使用确认机制。例如，在 main_team.zh.txt 中明确要求：

只有当历史状态已显示 sop_stage=await_outfit_confirmation，且用户明确表达"开始/开始生成/就按这个开始/生成吧"等含义时，才允许直接执行 run_outfit_flow。

Sources: descriptions.toml, descriptions.toml, main_team.zh.txt

测试用例与边界情况

用户确认机制通过全面的测试用例验证其正确性。测试覆盖以下关键场景：

1. 1. 状态验证测试：验证 run_outfit_flow 在未满足确认条件时返回错误
2. 2. 正常执行测试：验证工作流在正确确认后能够正常执行
3. 3. 状态更新测试：验证工作流完成后对话状态正确更新
4. 4. 中断恢复测试：验证工作流被中断后的状态恢复逻辑

例如，test_run_outfit_flow_requires_confirmation_state 测试验证了当 sop_stage 不是 await_outfit_confirmation 时，工具会返回 outfit_flow_not_ready_to_run 错误。

边界情况测试包括：

• - 工作流执行期间用户发送新消息
• - 用户在确认阶段补充信息而非确认开始
• - 工作流失败后的重试逻辑
• - 多个工作流实例的并发控制

这些测试确保了用户确认机制在各种边界情况下都能正确工作，防止系统进入不一致状态。

Sources: test_outfit_workflow_tools.py, test_dialog_state.py

与 SOP 阶段的集成

用户确认机制与 SOP 阶段紧密集成，形成完整的穿搭工作流控制链。SOP 阶段通过 sop_stage 字段标记当前流程位置：

在样式顾问技能（style_advisor_skill.zh.txt）中，明确要求在需求足够时立即设置确认状态：

需求足够时，立即调用 update_dialog_state 写入 outfit_task，并把 outfit_status="awaiting_confirmation"、sop_stage="await_outfit_confirmation"、pending_confirmation.action="run_outfit_flow" 一起写回。

这种集成确保了用户确认机制在 SOP 流程中的正确位置，既不会过早触发，也不会遗漏必要的确认步骤。

Sources: outfit_workflow_service.py, style_advisor_skill.zh.txt

最佳实践与开发指南

在使用用户确认机制时，开发者应遵循以下最佳实践：

1. 正确设置确认状态

# 正确示例：在需求收集完成后设置确认状态

update_dialog_state_fields(

    session_state,

    sop_stage="await_outfit_confirmation",

    outfit_task=task_payload,

    outfit_status="awaiting_confirmation",

    pending_confirmation={

        "action": "run_outfit_flow",

        "summary": "是否开始生成两套穿搭图",

    },

)

2. 验证确认状态

在执行敏感操作前，始终验证确认状态：

if not _is_outfit_confirmation_ready(

    dialog_state,

    allowed_actions={"run_outfit_flow"},

):

    return json.dumps({"error": "outfit_flow_not_ready_to_run"})

3. 处理中断情况

在工作流执行期间，正确处理用户中断：

try:

    result = await service.run_prepared_workflow(request=request, tool_ctx=ctx)

except asyncio.CancelledError:

    # 设置中断状态，等待用户确认恢复

    update_dialog_state_fields(

        session_state,

        sop_stage="await_outfit_confirmation",

        pending_confirmation={

            "action": "run_outfit_flow",

            "summary": "是否恢复中断的工作流",

        },

    )

    raise

4. 清理确认状态

工作流完成后，清理确认状态：

update_dialog_state_fields(

    session_state,

    clear_pending_confirmation=True,

)

5. 避免绕过确认

不要尝试通过其他工具绕过确认机制，系统会检测并阻止此类行为。

Sources: dialog_state.py, outfit_workflow_tools.py

架构图示

stateDiagram-v2

    [*] --> collecting_requirements: 开始对话

    collecting_requirements --> await_outfit_confirmation: 需求收集完成

    await_outfit_confirmation --> outfit_running: 用户确认开始

    await_outfit_confirmation --> collecting_requirements: 用户补充信息

    outfit_running --> s3_visual_delivered: 工作流完成

    outfit_running --> outfit_failed: 工作流失败

    outfit_running --> await_outfit_confirmation: 用户中断

    s3_visual_delivered --> [*]: 进入反馈阶段

    outfit_failed --> await_outfit_confirmation: 重新确认

    outfit_failed --> collecting_requirements: 重新收集需求

    

    state await_outfit_confirmation {

        [*] --> 等待用户确认

        等待用户确认 --> 确认开始: 用户说"开始生成"

        等待用户确认 --> 补充信息: 用户补充需求

    }

相关页面

• - Dialog State 设计：了解对话状态管理的完整设计
• - SOP 阶段推进：了解 SOP 阶段的完整流程
• - 工具注册与工厂：了解工具的注册和门控机制
• - 会话分发与回复状态：了解会话管理和中断处理

<a id="12-gong-ju-zhu-ce-yu-gong-han"></a>

---

工具注册与工厂

• - Section: 架构与核心概念
• - Group: 工具系统
• - Source File: 12-gong-ju-zhu-ce-yu-gong-han.md

本文档详细阐述 outfit_agent 系统中工具（Tool）的注册机制、工厂模式实现以及相关的安全门控体系。工具系统是连接 LLM 智能体与业务逻辑的核心桥梁，通过统一的注册表（Registry）和工厂模式（Factory Pattern）实现工具的标准化创建、动态加载和运行时访问控制。

核心架构：三层注册体系

outfit_agent 的工具系统采用三层注册架构，从底层到上层分别为：原始函数层、包装工具层、注册表层。这种分层设计确保了业务逻辑与 LLM 交互的安全隔离。

graph TB

    subgraph "原始函数层"

        A["业务函数<br/>(outfit_biz)"]

        B["工具包装函数<br/>(app/tools/*.py)"]

    end

    

    subgraph "包装工具层"

        C["build_tool()<br/>工厂函数"]

        D["Function 对象<br/>(agno.tools.Function)"]

    end

    

    subgraph "注册表层"

        E["TOOL_REGISTRY<br/>全局工具注册表"]

        F["SkillBundle<br/>技能工具包"]

    end

    

    A -->|"user_id, lang 等敏感参数<br/>由 ToolContext 注入"| B

    B -->|"纯业务意图参数<br/>(搜索词、ID 等)"| C

    C -->|"包装、追踪、门控"| D

    D -->|"按名称注册"| E

    E -->|"按技能声明分组"| F

    

    style A fill:#e1f5fe

    style B fill:#f3e5f5

    style C fill:#fff3e0

    style D fill:#e8f5e8

    style E fill:#fce4ec

    style F fill:#f1f8e9

Sources: app/tools/__init__.py, app/tools/factory.py

工具工厂模式详解

工具工厂（Tool Factory）是整个工具系统的核心，通过 build_tool() 函数实现从原始业务函数到 agno Function 对象的转换。

工厂函数签名

def build_tool(entrypoint: Callable[..., Any], *, name: str | None = None) -> Function

Sources: app/tools/factory.py

工厂处理流程

build_tool() 执行以下关键步骤：

1. 1. 名称解析：优先使用显式 name 参数，否则从函数名推导
2. 2. 入口包装：通过 _wrap_tool_entrypoint() 为函数添加安全上下文注入和追踪能力
3. 3. 描述加载：从 data/tools/descriptions.toml 加载工具描述，支持用户级灰度覆盖
4. 4. 行为配置：根据函数装饰器属性设置 stop_after_tool_call、show_result 等行为
5. 5. 钩子安装：为动态停止条件安装 pre_hook 和 post_hook

flowchart TD

    A["build_tool(entrypoint)"] --> B["解析工具名称"]

    B --> C["_wrap_tool_entrypoint()"]

    C --> D["Function.from_callable()"]

    D --> E["加载工具描述"]

    E --> F{"检查装饰器属性"}

    F -->|"__tool_stop_after_tool_call__"| G["设置 stop_after_tool_call"]

    F -->|"__tool_stop_after_tool_call_if__"| H["安装动态停止钩子"]

    F -->|"__tool_show_result__"| I["设置 show_result"]

    G --> J["返回 Function 对象"]

    H --> J

    I --> J

    

    style A fill:#fff3e0

    style J fill:#e8f5e8

Sources: app/tools/factory.py, app/tools/factory.py

工具描述管理

工具描述采用双层缓存机制，支持基础描述和用户级灰度覆盖：

Sources: app/tools/factory.py, data/tools/descriptions.toml

安全包装机制

RunContext 与 ToolContext

工具包装的核心安全目标是防止 LLM 越权访问用户数据。通过 RunContext 注入会话级上下文，确保 LLM 无法直接操作 user_id、lang 等敏感参数。

classDiagram

    class RunContext {

        +str run_id

        +str session_id

        +str user_id

        +dict session_state

        +dict dependencies

        +dict metadata

    }

    

    class ToolContext {

        +int user_id

        +str user_id_str

        +str lang

        +str session_id

        +str avatar_path

        +str avatar_info

        +str channel

        +dict channel_data

        +replace(**changes)

        +to_session_value()

    }

    

    class _wrap_tool_entrypoint {

        +_async_wrapper(*args, **kwargs)

        +_sync_wrapper(*args, **kwargs)

    }

    

    RunContext --> "session_state[tool_ctx]" ToolContext : 包含

    _wrap_tool_entrypoint --> RunContext : 读取

    _wrap_tool_entrypoint --> ToolContext : 注入业务函数

Sources: app/tools/context.py, app/tools/factory.py

包装函数的双重保障

_wrap_tool_entrypoint() 为每个工具函数添加以下安全机制：

1. 1. 上下文绑定：通过 bind_agno_context() 绑定 session_id、run_id、user_id 到追踪上下文
2. 2. 调用追踪：通过 trace_tool_call() 记录工具调用的输入、输出和属性
3. 3. 渠道门控：检查工具是否在当前渠道（channel）可用
4. 4. 技能门控：检查工具是否在当前激活的技能（skill）中声明

Sources: app/tools/factory.py, app/tools/factory.py

工具注册表（TOOL_REGISTRY）

注册表结构

TOOL_REGISTRY 是一个全局字典，映射工具名称到 Function 对象。它在应用启动时通过 init_tool_registry() 初始化。

TOOL_REGISTRY: dict[str, Function] = {}

Sources: app/tools/__init__.py

注册表构建流程

_build_tool_registry() 函数从各个工具模块收集原始函数，通过 build_tool() 转换为 Function 对象：

Sources: app/tools/__init__.py

工具分组常量

除了全局注册表，系统还维护多个工具分组常量，便于按业务域批量引用：

Sources: app/tools/__init__.py

动态工具加载机制

基于 Team 配置的加载

主 Agent 的工具集通过 data/team/main.toml 配置，支持全渠道基础工具和渠道专属工具：

# 全渠道基础工具

tools = [

    "run_outfit_flow",

    "update_user_profile",

    "new_session",

    "get_dialog_state",

    "update_dialog_state",

    "query_weather_now",

    "query_weather_forecast",

    "query_weather_hourly",

]



# 渠道专属工具

[channels.wechat_kefu]

tools = ["send_miniprogram_message"]

Sources: data/team/main.toml, app/agents/team_loader.py

工具解析与验证

resolve_tools_by_names() 函数从注册表中按名称解析工具，并验证工具存在性：

def resolve_tools_by_names(tool_names: Iterable[str], *, owner_id: str) -> list[Function]

Sources: app/tools/__init__.py

工具门控机制（Tool Gating）

门控架构

工具门控是运行时访问控制的核心，确保工具只能在正确的技能上下文中被调用。

flowchart TD

    A["工具调用请求"] --> B{"工具在 ALWAYS_ALLOWED_TOOLS 中?"}

    B -->|"是"| C["允许调用"]

    B -->|"否"| D{"工具有声明所属技能?"}

    D -->|"否"| C

    D -->|"是"| E{"当前激活技能匹配?"}

    E -->|"是"| C

    E -->|"否"| F["返回门控错误"]

    

    style C fill:#e8f5e8

    style F fill:#ffebee

Sources: app/dialog_state.py, app/dialog_state.py

常驻允许工具

以下工具不受门控限制，始终可用：

• - run_outfit_flow - 穿搭工作流
• - update_user_profile - 用户档案更新
• - new_session - 新会话
• - send_text_message - 文本消息
• - send_miniprogram_message - 小程序消息
• - get_skill_instructions - 技能指令获取
• - get_skill_reference - 技能参考获取
• - get_skill_script - 技能脚本获取
• - get_dialog_state - 对话状态获取
• - update_dialog_state - 对话状态更新

Sources: app/dialog_state.py

渠道门控

除了技能门控，系统还支持渠道门控，限制工具在特定渠道（如微信客服、小程序）中可用：

def _tool_available_for_current_channel(run_context: Any, tool_name: str) -> str | None

Sources: app/tools/factory.py

动态停止条件

条件停止钩子

某些工具（如 run_outfit_flow）需要根据返回结果动态决定是否停止 Agent 运行。通过 _install_dynamic_stop_after_hook() 实现：

def _install_dynamic_stop_after_hook(

    tool: Function,

    *,

    condition: Callable[[Any], bool],

) -> None

工作原理：

1. 1. pre_hook：每次调用前重置 stop_after_tool_call = False
2. 2. post_hook：调用后根据 condition(result) 动态设置 stop_after_tool_call

Sources: app/tools/factory.py, tests/test_tool_factory.py

技能工具绑定

技能声明工具

每个技能在 skill.toml 中声明自己拥有的工具：

[skill]

tool_names = ["search_items", "get_item_detail", ...]

Sources: app/skills/loader.py

技能工具包（SkillBundle）

当技能被激活时，系统构建 SkillBundle，包含技能声明的工具和访问工具：

@dataclass(frozen=True)

class SkillBundle:

    names: list[str]

    tools: list[Any]

    access_tools: list[Any]

    prompt_snippet: str

    always_instruction_prompts: list[str]

    inline_instructions: str

    skills: object | None

    definitions: list[RegisteredSkill]

Sources: app/skills/loader.py

工具调用追踪

追踪属性构建

每次工具调用都会记录以下追踪属性：

def _build_trace_attributes(run_context: Any) -> dict[str, Any]:

    return {

        "skill.id": dialog_state.get("active_skill"),

        "scene.code": dialog_state.get("active_scene"),

        "sop.stage": dialog_state.get("sop_stage"),

        "workflow.status": dialog_state.get("outfit_status"),

        "workflow.run_id": dialog_state.get("outfit_run_id"),

        "dialog.awaiting_confirmation": bool(pending_confirmation),

        "dialog.loaded_skills": dialog_state.get("loaded_skills") or [],

    }

Sources: app/tools/factory.py

最佳实践

新增工具的步骤

1. 1. 创建工具函数：在 app/tools/ 下对应模块中创建业务函数
2. 2. 添加包装函数：在函数签名中添加 run_context: RunContext 参数，通过 _get_ctx() 获取上下文
3. 3. 注册到 TOOLS 列表：在模块末尾的 TOOLS 列表中添加函数引用
4. 4. 更新注册表：在 app/tools/__init__.py 的 _build_tool_registry() 中添加映射
5. 5. 添加工具描述：在 data/tools/descriptions.toml 中添加工具描述
6. 6. 配置技能声明：在相应技能的 skill.toml 中声明工具

安全注意事项

• - 永远不要在工具函数签名中暴露 user_id、lang 等敏感参数
• - 始终使用 _get_ctx(run_context) 获取用户上下文
• - 验证权限：在访问用户数据前检查 ctx.user_id 是否匹配
• - 处理缺失上下文：当 _get_ctx() 返回 None 时返回 _NO_CTX_RESULT

Sources: app/tools/context.py, app/tools/item_tools.py

相关页面

• - 工具描述管理 - 了解工具描述的加载、缓存和灰度机制
• - 工具调用链路追踪 - 了解工具调用的追踪和监控机制
• - 技能工具门控机制 - 了解技能与工具的绑定和访问控制
• - Dialog State 设计 - 了解对话状态管理，这是工具门控的基础
• - 新增业务工具 - 了解如何扩展新的业务工具

<a id="13-gong-ju-miao-shu-guan-li"></a>

---

工具描述管理

• - Section: 架构与核心概念
• - Group: 工具系统
• - Source File: 13-gong-ju-miao-shu-guan-li.md

工具描述管理是 Outfit Agent 系统中用于控制 LLM 工具行为的核心机制。通过集中化的描述文件和灵活的加载策略，系统能够精确指导 LLM 如何使用每个工具，同时支持灰色发布以进行 A/B 测试和用户级别的定制化调整。

工具描述文件结构与格式

所有工具描述集中存储在 data/tools/descriptions.toml 文件中，采用 TOML 格式。每个工具在 [tools] 表下有一个键值对，其中键为工具名称，值为该工具的多行描述字符串。

文件结构示例：

[tools]

generate_image = '''

适合用户想修改已有穿搭图或上一张 tryon 试穿图时调用。

根据文字提示词直接生成 1 张图片，可选附带 0～9 张参考图，输出比例固定为 3:4。

必须提供 `prompt`；可选传 `ref_images`，其中每一项都应是系统里已有的 `image_id`，总数最多 9 张。

本工具成功后会自动发送生成图片；

返回 JSON 对象，通常包含生成结果的 `image_id`、`oss_url` 和发送结果。

'''



search_items = '''

在公共商品库中按文本搜索单品。

适合当你已经有明确的文字检索条件，需要为用户找新商品、补齐穿搭缺失单品，或按款式/颜色/材质/场景查找候选单品时调用。

如果同一轮里要同时搜索多件不同单品，优先使用 `search_items_batch`，把所有请求一次性提交，减少重复 embedding 和多次串行搜品。

传入的 `query` 建议优先整理成英文关键词，便于和单品 VLM 英文字段、BM25 检索保持一致；

可选传入 `product_type`、`sub_category`、`leaf_category` 缩小品类，传入 `image_id` 做图文联合检索，并用 `limit` 控制结果数。

若传分类参数，必须使用精确 taxonomy 值；系统会优先尝试三级分类过滤，命中数不足时自动回退到二级/一级分类。

如果用户已经给了参考图，且目标是“找图中单品 / 找同款 / 找相似款”，不要把本工具当默认入口；那种场景应优先使用 `search_items_by_image_tool`。

这里的 `image_id` 更适合“文字检索为主、图片仅辅助排序”的混合检索，不适合作为参考图找同款的主流程。

不要在仅需搜索用户自有衣橱时调用本工具；那种场景应使用 `search_wardrobe`。

对于"搜同款发给我/给我看看候选"这类场景，拿到多个候选后通常应继续调用 `send_items` 做视觉展示，而不是只返回长段文字。

返回 JSON 数组，每项通常包含 `item_id`、`product_type`、`sub_category`、`leaf_category`、`brand`、`description`、`image_id`、`price` 等字段。

'''

描述编写规范：

1. 1. 场景说明：明确描述何时使用该工具，避免误用
2. 2. 参数说明：详细说明必选和可选参数，包括格式要求
3. 3. 行为边界：明确工具的使用限制和替代方案
4. 4. 返回格式：说明返回的数据结构和字段含义
5. 5. 注意事项：包含重要的警告和最佳实践

Sources: data/tools/descriptions.toml

工具描述加载机制

系统采用分层缓存策略加载工具描述，确保性能的同时支持灰色发布。

加载流程：

1. 1. 基础描述加载：从 data/tools/descriptions.toml 文件读取基础描述
2. 2. 用户覆盖加载：检查是否有针对当前用户的灰色发布描述
3. 3. 描述合并：将用户特定描述与基础描述合并
4. 4. 缓存管理：使用 LRU 缓存避免重复文件读取

关键函数：

@lru_cache(maxsize=1)

def _load_base_tool_descriptions() -> dict[str, str]:

    """加载基础工具描述"""

    content = get_file("tools/descriptions.toml")

    data = tomllib.loads(content)

    return dict(data["tools"])



@lru_cache(maxsize=128)

def _load_tool_descriptions(user_id: str | None = None) -> dict[str, str]:

    """加载工具描述，支持用户级覆盖"""

    if user_id is None:

        return _load_base_tool_descriptions()

    

    # 加载用户特定描述

    content = get_file("tools/descriptions.toml", user_id=user_id)

    data = tomllib.loads(content)

    overlay = data["tools"]

    

    # 合并描述

    return {

        **_load_base_tool_descriptions(),

        **{str(key): str(value) for key, value in overlay.items()},

    }

缓存策略：

• - 基础描述：全局缓存，应用生命周期内只加载一次
• - 用户描述：LRU 缓存，最多缓存 128 个用户描述
• - 缓存清除：通过 clear_tool_description_cache() 函数手动清除

Sources: app/tools/factory.py

工具描述应用流程

工具描述在工具构建和用户会话初始化时应用到工具对象。

工具构建阶段：

def build_tool(entrypoint: Callable[..., Any], *, name: str | None = None) -> Function:

    """构建工具对象并应用描述"""

    tool_name = name or entrypoint.__name__

    wrapped = _wrap_tool_entrypoint(entrypoint, tool_name=tool_name)

    tool = Function.from_callable(wrapped, name=tool_name)

    

    # 应用工具描述

    tool.description = get_tool_description(tool_name)

    

    # 处理其他属性

    stop_after_tool_call = bool(getattr(entrypoint, "__tool_stop_after_tool_call__", False))

    # ...

    return tool

用户会话阶段：

def materialize_tools_for_user(

    tools: Sequence[Function | Callable[..., Any]],

    *,

    user_id: int | str | None = None,

) -> list[Function | Callable[..., Any]]:

    """为特定用户物化工具描述"""

    resolved_user_id = _normalize_tool_description_user_id(user_id)

    descriptions = _load_tool_descriptions(resolved_user_id)

    

    materialized: list[Function | Callable[..., Any]] = []

    for tool in tools:

        if not isinstance(tool, Function):

            materialized.append(tool)

            continue

        

        # 深拷贝工具对象避免影响原始工具

        copied = tool.model_copy(deep=True)

        description = descriptions.get(copied.name)

        

        # 应用用户特定描述

        if isinstance(description, str) and description.strip():

            copied.description = description

        

        materialized.append(copied)

    return materialized

工具注册流程：

# 1. 在 app/tools/__init__.py 中初始化工具注册表

TOOL_REGISTRY: dict[str, Function] = {}



def _build_tool_registry() -> dict:

    """构建工具注册表"""

    raw_registry = {

        "search_items": search_items,

        "search_items_batch": search_items_batch,

        # ... 其他工具

    }

    return {name: build_tool(fn, name=name) for name, fn in raw_registry.items()}



# 2. 启动时初始化注册表

def init_tool_registry() -> None:

    """初始化工具注册表"""

    global TOOL_REGISTRY

    TOOL_REGISTRY.clear()

    TOOL_REGISTRY.update(_build_tool_registry())

Sources: app/tools/factory.py, app/tools/__init__.py

灰色发布机制

系统支持基于用户的工具描述灰色发布，用于 A/B 测试、功能渐进式发布或用户个性化调整。

灰色发布架构：

graph TB

    A[用户请求] --> B{是否有用户ID?}

    B -->|是| C[加载用户特定描述]

    B -->|否| D[加载基础描述]

    C --> E[合并描述]

    D --> F[直接使用基础描述]

    E --> G[应用到工具对象]

    F --> G

实现细节：

1. 1. 描述覆盖：用户特定描述文件结构与基础文件相同，但只包含需要覆盖的工具
2. 2. 合并策略：用户特定描述完全覆盖基础描述中的同名工具
3. 3. 缓存机制：每个用户的描述独立缓存，避免相互影响
4. 4. 性能优化：使用 LRU 缓存，最多缓存 128 个用户的描述

使用场景：

• - A/B 测试：测试不同工具描述对 LLM 行为的影响
• - 功能渐进式发布：先对特定用户发布新工具描述，观察效果后再全量发布
• - 用户个性化：根据用户偏好调整工具描述
• - 错误修复：快速修复特定用户的工具描述问题

管理接口： 灰色发布通过 Admin API 管理，具体接口在 app/routers/admin/gray.py 中定义：

• - POST /admin/gray/rules：创建灰色发布规则
• - PUT /admin/gray/rules/{rule_id}：更新灰色发布规则
• - DELETE /admin/gray/rules/{rule_id}：删除灰色发布规则
• - POST /admin/gray/preview：预览灰色发布效果

Sources: app/routers/admin/gray.py, tests/test_tool_description_gray.py

工具描述管理接口

系统提供完整的 Admin API 用于管理工具描述文件。

核心接口：

热重载机制：

@router.post("/reload", summary="热重载工具描述到运行时")

async def reload_tool_descriptions(_: AdminRequired) -> JSONResponse:

    """热重载工具描述"""

    try:

        from app.agent import clear_runtime_tool_cache, reload_agents

        from app.tools.factory import clear_tool_description_cache, list_tool_descriptions

        

        # 清除缓存

        clear_tool_description_cache()

        clear_runtime_tool_cache()

        

        # 重新加载代理

        t0 = time.monotonic()

        reload_agents()

        elapsed_ms = int((time.monotonic() - t0) * 1000)

        

        # 获取工具列表

        tool_names = sorted(list_tool_descriptions().keys())

        

        return JSONResponse({

            "ok": True,

            "tool_names": tool_names,

            "tool_count": len(tool_names),

            "elapsed_ms": elapsed_ms,

        })

    except Exception as exc:

        logger.error("Admin tool description reload failed: %s", exc, exc_info=True)

        raise HTTPException(status_code=500, detail=str(exc))

版本管理：

• - 自动备份：每次更新前自动备份当前版本
• - 历史版本：保留最近 10 个历史版本
• - 版本回滚：支持回滚到任意历史版本
• - 版本验证：回滚前验证内容格式的正确性

Sources: app/routers/admin/tool_descriptions.py

与技能系统的集成

工具描述与技能系统紧密集成，确保工具在正确的技能上下文中使用。

技能工具声明： 在 skill.toml 文件中，技能通过 tools 字段声明其拥有的工具：

# data/skills/buyer-wardrobe-skill/skill.toml

id = "buyer-wardrobe-skill"

name = "Buyer Wardrobe Skill"

description = "买手衣橱技能"

activation = "on_demand"



[metadata]

category = "business"

owner = "main-team"



[instructions]

zh = "skills/buyer_wardrobe.zh.txt"

en = "skills/buyer_wardrobe.en.txt"



# 声明技能拥有的工具

tools = [

    "search_items",

    "search_wardrobe",

    "search_items_by_image_tool",

    # ... 其他工具

]

工具门控机制： 系统通过工具门控机制确保工具只在正确的技能上下文中使用：

# app/dialog_state.py

def is_tool_allowed(

    session_state: dict[str, Any] | None,

    *,

    tool_name: str,

    owning_skills: list[str],

) -> bool:

    """检查工具是否在当前技能上下文中允许使用"""

    dialog_state = get_dialog_state(session_state)

    active_skill = dialog_state.get("active_skill")

    

    # 如果没有活动技能，只允许常驻工具

    if not active_skill:

        return tool_name in _RESIDENT_TOOLS

    

    # 如果工具属于活动技能，允许使用

    if active_skill in owning_skills:

        return True

    

    # 如果工具不属于任何技能，作为常驻工具允许

    if not owning_skills:

        return True

    

    return False

工具描述应用时机：

1. 1. 技能激活时：当用户激活特定技能时，系统加载该技能的工具描述
2. 2. 工具调用时：在工具调用前，系统检查工具是否在当前技能上下文中允许
3. 3. 描述覆盖：技能可以覆盖基础工具描述，提供更具体的上下文指导

工具描述与技能的协同：

• - 技能上下文：工具描述应与技能的业务上下文保持一致
• - 参数指导：技能可以提供更具体的参数使用指导
• - 行为约束：技能可以约束工具在特定场景下的行为
• - 错误处理：技能可以定义特定场景下的错误处理策略

Sources: app/dialog_state.py, app/skills/loader.py

最佳实践

工具描述编写指南

1. 1. 场景明确：清晰描述工具的使用场景，避免误用
2. 2. 参数详细：详细说明必选和可选参数，包括格式要求和取值范围
3. 3. 行为边界：明确工具的使用限制和替代方案
4. 4. 返回格式：说明返回的数据结构和字段含义
5. 5. 注意事项：包含重要的警告和最佳实践

灰色发布策略

1. 1. 小范围测试：先对小范围用户测试新工具描述
2. 2. 效果监控：监控工具描述变更对 LLM 行为的影响
3. 3. 快速回滚：准备快速回滚机制，应对问题描述
4. 4. 逐步发布：通过灰色发布逐步扩大新描述的适用范围

性能优化

1. 1. 缓存利用：充分利用 LRU 缓存，避免重复文件读取
2. 2. 延迟加载：只在需要时加载用户特定描述
3. 3. 缓存清理：及时清理无效缓存，确保数据一致性
4. 4. 监控指标：监控工具描述加载的性能指标

安全考虑

1. 1. 输入验证：验证工具描述内容的格式和结构
2. 2. 权限控制：限制工具描述的修改权限
3. 3. 审计日志：记录工具描述的修改和访问日志
4. 4. 备份策略：制定工具描述文件的备份策略

Sources: data/tools/descriptions.toml, app/tools/factory.py, app/routers/admin/tool_descriptions.py

下一步阅读

了解工具描述管理后，建议继续阅读以下相关文档：

• - 工具注册与工厂 - 了解工具如何注册到系统
• - 工具调用链路追踪 - 了解工具调用的监控和调试
• - 技能加载与管理 - 了解技能系统如何与工具集成
• - 技能工具门控机制 - 了解工具使用的权限控制机制

<a id="14-gong-ju-diao-yong-lian-lu-zhui-zong"></a>

---

工具调用链路追踪

• - Section: 架构与核心概念
• - Group: 工具系统
• - Source File: 14-gong-ju-diao-yong-lian-lu-zhui-zong.md

在复杂的 Agent 系统中，工具调用链路追踪是保障系统可观测性、调试效率和性能优化的关键基础设施。本文档详细介绍了 outfit_agent 中工具调用链路追踪的架构设计、核心组件和实现机制，帮助开发者理解如何追踪、监控和分析工具调用链路。

架构概览

工具调用链路追踪系统采用分层架构，从底层的 OpenTelemetry 集成到上层的实时监控和可视化，形成了完整的追踪链路。

graph TB

    subgraph "应用层"

        A[Agent 工具调用] --> B[工具包装层]

        B --> C[追踪上下文绑定]

        C --> D[追踪数据收集]

    end

    

    subgraph "追踪基础设施"

        E[OpenTelemetry SDK]

        F[Agno 追踪集成]

        G[HyperDX 日志聚合]

    end

    

    subgraph "存储与查询"

        H[PostgreSQL 追踪存储]

        I[Redis 实时状态]

        J[Admin API]

    end

    

    subgraph "可视化层"

        K[Dashboard 追踪面板]

        L[实时运行监控]

    end

    

    A --> E

    E --> F

    F --> H

    D --> I

    J --> K

    J --> L

    

    style A fill:#e1f5fe

    style E fill:#f3e5f5

    style H fill:#e8f5e8

    style K fill:#fff3e0

Sources: app/agentos.py, app/observability.py

核心组件

1. 追踪上下文绑定

追踪上下文绑定是工具调用链路追踪的基础，它确保每次工具调用都关联到正确的会话、运行和用户上下文。

# 追踪上下文绑定示例

from outfit_biz.utils.agno_trace import bind_agno_context



with bind_agno_context(

    session_id="session-123",

    run_id="run-456", 

    user_id="user-789",

    tool_name="search_items"

):

    # 工具执行逻辑

    pass

关键特性：

• - 会话级隔离：每个会话拥有独立的追踪上下文
• - 运行时绑定：上下文在工具调用时动态绑定
• - 自动清理：工具调用完成后自动清理上下文

Sources: app/tools/factory.py, app/img_gen_handler.py

2. 工具调用追踪

工具调用追踪通过 trace_tool_call 函数实现，它为每次工具调用创建详细的追踪记录。

# 工具调用追踪示例

from outfit_biz.utils.agno_trace import trace_tool_call



with trace_tool_call(

    "search_items",

    attributes={

        "skill.id": "buyer-wardrobe-skill",

        "scene.code": "outfit_recommendation",

        "sop.stage": "await_search_confirmation",

        "outfit.tool.source": "agent_tool",

    },

    input_value={"query": "夏季连衣裙"},

    session_id="session-123",

    run_id="run-456",

    user_id="user-789"

) as traced:

    # 执行工具逻辑

    result = await search_items(query="夏季连衣裙")

    traced.set_output(result)

追踪属性：

• - 技能信息：skill.id、scene.code、sop.stage
• - 工作流状态：workflow.status、workflow.run_id
• - 对话状态：dialog.awaiting_confirmation、dialog.loaded_skills
• - 工具来源：outfit.tool.source（agent_tool/internal）

Sources: app/tools/factory.py, app/workflows/outfit_generation.py

3. 实时运行监控

实时运行监控通过 Redis 存储运行状态，提供工具调用的实时追踪能力。

sequenceDiagram

    participant Agent as Agent 运行时

    participant Hook as 工具钩子

    participant Redis as Redis 存储

    participant API as Admin API

    participant Dashboard as 监控面板

    

    Agent->>Hook: 工具调用开始

    Hook->>Redis: 注册工具调用状态

    Redis-->>API: 提供实时数据

    API-->>Dashboard: 展示运行状态

    

    Agent->>Hook: 工具调用完成

    Hook->>Redis: 更新工具调用结果

    Redis-->>API: 更新实时数据

    API-->>Dashboard: 更新监控面板

核心功能：

• - 工具调用生命周期管理：_start_tool_call 和 _finish_tool_call
• - 状态快照：记录工具调用的输入、输出和状态
• - 心跳机制：通过心跳保持运行状态活跃
• - 自动清理：TTL 机制自动清理过期数据

Sources: app/live_runs.py, app/agent.py

实现细节

1. 工具包装层

工具包装层是追踪系统的核心，它为每个工具函数添加追踪和上下文绑定能力。

# 工具包装函数示例

def _wrap_tool_entrypoint(entrypoint, *, tool_name):

    @wraps(entrypoint)

    async def _async_wrapper(*args, **kwargs):

        run_context = kwargs.get("run_context")

        

        # 绑定追踪上下文

        with bind_agno_context(

            session_id=getattr(run_context, "session_id", None),

            run_id=getattr(run_context, "run_id", None),

            user_id=_resolve_tool_user_id(run_context),

            tool_name=tool_name,

        ):

            # 创建追踪记录

            with trace_tool_call(

                tool_name,

                attributes=_build_trace_attributes(run_context),

                input_value={"args_count": len(args)},

                session_id=getattr(run_context, "session_id", None),

                run_id=getattr(run_context, "run_id", None),

                user_id=_resolve_tool_user_id(run_context),

            ) as traced:

                # 执行门控检查

                if not is_tool_allowed(...):

                    result = build_tool_gate_error(...)

                    traced.set_output(result)

                    return result

                

                # 执行工具逻辑

                result = await entrypoint(*args, **kwargs)

                traced.set_output(result)

                return result

    

    return _async_wrapper

包装层职责：

1. 1. 上下文提取：从 RunContext 提取会话、运行和用户信息
2. 2. 追踪绑定：绑定 OpenTelemetry 追踪上下文
3. 3. 门控检查：执行工具访问权限和渠道限制检查
4. 4. 结果记录：记录工具调用的输入和输出

Sources: app/tools/factory.py

2. 追踪属性构建

追踪属性构建函数 _build_trace_attributes 从对话状态中提取关键信息，为追踪记录提供丰富的上下文。

def _build_trace_attributes(run_context):

    session_state = getattr(run_context, "session_state", None)

    dialog_state = get_dialog_state(session_state)

    pending_confirmation = dialog_state.get("pending_confirmation")

    

    return {

        "skill.id": dialog_state.get("active_skill"),

        "scene.code": dialog_state.get("active_scene"),

        "sop.stage": dialog_state.get("sop_stage"),

        "workflow.status": dialog_state.get("outfit_status"),

        "workflow.run_id": dialog_state.get("outfit_run_id"),

        "dialog.awaiting_confirmation": bool(pending_confirmation),

        "dialog.loaded_skills": dialog_state.get("loaded_skills") or [],

    }

属性分类：

• - 技能上下文：当前激活的技能、场景和 SOP 阶段
• - 工作流状态：穿搭生成工作流的状态和运行 ID
• - 对话状态：是否等待确认、已加载的技能列表

Sources: app/tools/factory.py

3. LLM 请求日志

LLM 请求日志通过 LoggedOpenAIChat 类实现，记录详细的 LLM 调用信息。

# LLM 请求日志示例

class LoggedOpenAIChat(OpenAIChat):

    async def ainvoke(self, messages, assistant_message, **kwargs):

        # 构建请求快照

        snapshot = self._build_request_snapshot(

            messages=messages,

            response_format=response_format,

            tools=tools,

            tool_choice=tool_choice,

            run_response=run_response,

        )

        

        # 记录请求开始

        self._log_request_start("async", snapshot, client_summary)

        

        # 执行请求

        response = await super().ainvoke(...)

        

        # 记录请求完成

        self._log_request_end("async", snapshot, elapsed_ms, response)

        

        return response

日志内容：

• - 请求摘要：消息数量、角色分布、内容大小
• - 模型信息：模型 ID、基础 URL、超时设置
• - 性能指标：请求耗时、响应大小
• - 错误信息：请求失败时的详细错误

Sources: app/llm_logging.py

配置与使用

1. 环境变量配置

追踪系统依赖以下环境变量进行配置：

# HyperDX 监控配置

HYPERDX_API_KEY=your_hyperdx_api_key

OTEL_SERVICE_NAME=outfit-agent

OTEL_EXPORTER_OTLP_ENDPOINT=https://in-otel.hyperdx.io



# Redis 配置（用于实时运行监控）

REDIS_URL=redis://localhost:6379/0



# 数据库配置（用于追踪存储）

DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/outfit

Sources: .env.example, config.yml

2. 追踪系统初始化

追踪系统在应用启动时自动初始化：

# 追踪系统初始化流程

def init_hyperdx_monitoring():

    config = load_hyperdx_config()

    if config is None:

        return False

    

    # 初始化日志导出器

    _ensure_log_exporter(config)

    

    logger.info("HyperDX monitoring enabled for logs only")

    return True



# AgentOS 追踪初始化

def ensure_agno_db_tracing(db):

    provider = trace_api.get_tracer_provider()

    

    # 添加数据库跨度导出器

    provider.add_span_processor(

        BatchSpanProcessor(DatabaseSpanExporter(db=db))

    )

    

    # 初始化 Agno 仪表化

    instrumentor = AgnoInstrumentor()

    instrumentor.instrument(tracer_provider=provider)

Sources: app/observability.py, app/agentos.py

3. 工具注册与追踪

工具注册时自动添加追踪包装：

# 工具注册示例

def build_tool(entrypoint, *, name=None):

    tool_name = name or entrypoint.__name__

    

    # 包装工具入口点

    wrapped = _wrap_tool_entrypoint(entrypoint, tool_name=tool_name)

    

    # 创建工具函数

    tool = Function.from_callable(wrapped, name=tool_name)

    tool.description = get_tool_description(tool_name)

    

    # 设置工具属性

    if getattr(entrypoint, "__tool_stop_after_tool_call__", False):

        tool.stop_after_tool_call = True

        tool.show_result = True

    

    return tool

Sources: app/tools/factory.py

监控与调试

1. Admin API 接口

Admin API 提供丰富的追踪查询接口：

查询参数示例：

# 按用户 ID 过滤会话

GET /traces/sessions?user_id=42&limit=20&page=1



# 获取运行详情（包含基础设施跨度）

GET /traces/runs/run-123?include_infra_spans=true



# 获取实时运行列表

GET /traces/live-runs?session_id=session-456&limit=50

Sources: app/routers/admin/traces.py

2. Dashboard 追踪面板

Dashboard 提供可视化的追踪面板，支持以下功能：

会话列表视图：

• - 会话 ID、用户 ID、创建时间
• - 追踪数量、错误数量、最后追踪时间
• - Agent ID、Team ID、渠道信息
• - 灰度配置摘要

运行详情视图：

• - 运行状态、耗时、Agent 信息
• - 工具调用列表（名称、状态、耗时）
• - LLM 请求摘要（消息数量、Token 使用）
• - 对话状态快照（技能、场景、SOP 阶段）

实时运行监控：

• - 当前运行状态、工具调用状态
• - 实时更新（通过轮询或 WebSocket）
• - 工具调用输入/输出预览
• - 性能指标和错误信息

Sources: dashboard/src/pages/TraceInspector.tsx, dashboard/src/api/traces.ts

3. 追踪数据查询示例

查询用户会话历史：

SELECT 

    s.session_id,

    s.user_id,

    s.created_at,

    s.updated_at,

    COUNT(t.trace_id) as trace_count,

    COUNT(CASE WHEN t.status = 'error' THEN 1 END) as error_count

FROM agno.agno_sessions s

LEFT JOIN agno.agno_traces t ON s.session_id = t.session_id

WHERE s.user_id = '42'

GROUP BY s.session_id, s.user_id, s.created_at, s.updated_at

ORDER BY s.updated_at DESC

LIMIT 20;

查询工具调用详情：

SELECT 

    sp.span_id,

    sp.name as tool_name,

    sp.attributes->>'outfit.tool.source' as tool_source,

    sp.attributes->>'skill.id' as skill_id,

    sp.attributes->>'sop.stage' as sop_stage,

    sp.start_time,

    sp.end_time,

    EXTRACT(EPOCH FROM (sp.end_time - sp.start_time)) * 1000 as duration_ms

FROM agno.agno_spans sp

WHERE sp.trace_id = 'trace-123'

    AND sp.span_kind = 'tool'

ORDER BY sp.start_time;

最佳实践

1. 工具开发规范

必须遵循的规范：

1. 1. 使用工具包装：所有工具必须通过 build_tool 函数注册
2. 2. 提供清晰描述：在 tools/descriptions.toml 中为每个工具提供描述
3. 3. 处理错误：工具内部错误应被捕获并记录到追踪系统
4. 4. 设置超时：为长时间运行的工具设置合理的超时时间

工具描述示例：

# tools/descriptions.toml

[tools]

search_items = "根据关键词搜索商品，支持多维度筛选"

generate_outfit = "生成穿搭方案，包含图片和描述"

send_text_message = "发送文本消息给用户"

Sources: data/tools/descriptions.toml

2. 追踪属性使用

推荐使用的追踪属性：

• - skill.id：当前激活的技能 ID
• - scene.code：当前场景代码
• - sop.stage：SOP 阶段
• - outfit.tool.source：工具来源（agent_tool/internal）
• - workflow.status：工作流状态
• - dialog.awaiting_confirmation：是否等待确认

避免的属性：

• - 敏感信息：用户密码、Token、密钥
• - 大量数据：完整的请求/响应体
• - 动态变化：频繁变化的计数器

3. 性能优化建议

追踪性能优化：

1. 1. 采样率控制：在生产环境设置合理的采样率
2. 2. 批量处理：使用批量处理器减少数据库写入
3. 3. 异步导出：确保追踪导出不阻塞主业务流程
4. 4. 索引优化：为常用查询字段创建数据库索引

内存使用优化：

1. 1. 及时清理：确保追踪上下文及时清理
2. 2. 限制大小：限制单个追踪记录的大小
3. 3. 压缩存储：对大型追踪数据使用压缩存储

故障排查

1. 常见问题排查

追踪数据丢失：

1. 1. 检查环境变量配置：HYPERDX_API_KEY、OTEL_SERVICE_NAME
2. 2. 检查数据库连接：确保 PostgreSQL 连接正常
3. 3. 检查 Redis 连接：确保 Redis 连接正常
4. 4. 查看应用日志：检查追踪初始化错误

实时监控延迟：

1. 1. 检查 Redis 性能：确保 Redis 响应时间正常
2. 2. 检查心跳间隔：调整 live_run_heartbeat_interval_seconds
3. 3. 检查 TTL 设置：调整 live_run_redis_ttl_seconds

工具调用未追踪：

1. 1. 检查工具注册：确保工具通过 build_tool 注册
2. 2. 检查包装层：确保 _wrap_tool_entrypoint 正常工作
3. 3. 检查门控逻辑：确保工具未被门控拦截

2. 调试工具

查看追踪上下文：

from outfit_biz.utils.agno_trace import get_agno_trace_context



# 获取当前追踪上下文

ctx = get_agno_trace_context()

print(f"Session ID: {ctx.session_id}")

print(f"Run ID: {ctx.run_id}")

print(f"User ID: {ctx.user_id}")

print(f"Tool Name: {ctx.tool_name}")

检查工具注册：

from app.tools.factory import has_tool, get_tool_name



# 检查工具是否注册

tools = get_all_tools()

print(f"Tool 'search_items' registered: {has_tool(tools, 'search_items')}")



# 获取工具名称

for tool in tools:

    print(f"Tool: {get_tool_name(tool)}")

查看实时运行状态：

from app.live_runs import list_live_run_summaries



# 获取实时运行列表

runs = await list_live_run_summaries(session_id="session-123")

for run in runs:

    print(f"Run ID: {run['run_id']}")

    print(f"Status: {run['status']}")

    print(f"Active Tools: {run['active_tool_names']}")

扩展与定制

1. 自定义追踪属性

可以通过扩展 _build_trace_attributes 函数添加自定义追踪属性：

def _build_trace_attributes(run_context):

    # 原有属性

    attributes = {

        "skill.id": dialog_state.get("active_skill"),

        "scene.code": dialog_state.get("active_scene"),

        # ... 其他属性

    }

    

    # 自定义属性

    attributes.update({

        "custom.business_type": "outfit_recommendation",

        "custom.user_segment": "premium",

        "custom.ab_test_group": "group_a",

    })

    

    return attributes

2. 集成第三方监控

可以扩展追踪系统以集成第三方监控平台：

# 集成 Jaeger

from opentelemetry.exporter.jaeger import JaegerExporter



jaeger_exporter = JaegerExporter(

    agent_host_name="localhost",

    agent_port=6831,

)



provider.add_span_processor(

    BatchSpanProcessor(jaeger_exporter)

)



# 集成 Zipkin

from opentelemetry.exporter.zipkin import ZipkinExporter



zipkin_exporter = ZipkinExporter(

    endpoint="http://localhost:9411/api/v2/spans",

)



provider.add_span_processor(

    BatchSpanProcessor(zipkin_exporter)

)

3. 自定义 Dashboard 组件

可以扩展 Dashboard 以添加自定义追踪组件：

// 自定义追踪面板组件

const CustomTracePanel = ({ sessionId }) => {

  const [traces, setTraces] = useState([]);

  

  useEffect(() => {

    // 加载追踪数据

    tracesApi.getSession(sessionId).then(data => {

      setTraces(data.runs);

    });

  }, [sessionId]);

  

  return (

    <div>

      <h3>自定义追踪面板</h3>

      {traces.map(trace => (

        <div key={trace.run_id}>

          <span>{trace.run_id}</span>

          <span>{trace.status}</span>

          <span>{trace.duration_ms}ms</span>

        </div>

      ))}

    </div>

  );

};

总结

工具调用链路追踪系统是 outfit_agent 可观测性的核心组成部分，它通过多层次的追踪机制，提供了从工具调用到 LLM 请求的完整链路追踪能力。系统的主要特点包括：

1. 1. 完整的追踪链路：从工具调用到 LLM 请求的完整追踪
2. 2. 实时监控能力：基于 Redis 的实时运行状态监控
3. 3. 丰富的查询接口：Admin API 提供多维度的追踪查询
4. 4. 可视化支持：Dashboard 提供直观的追踪面板
5. 5. 灵活的扩展性：支持自定义追踪属性和第三方集成

通过合理使用追踪系统，开发者可以快速定位性能瓶颈、调试工具调用问题，并优化 Agent 系统的整体性能。

下一步

了解工具调用链路追踪后，建议继续阅读：

• - 工具注册与工厂 - 了解工具如何注册和管理
• - 工具描述管理 - 了解工具描述的配置和管理
• - 链路追踪与 Trace - 了解更广泛的追踪系统架构
• - 系统监控与健康检查 - 了解系统监控和健康检查机制

<a id="15-xiao-xi-ru-zhan-chu-li-liu-cheng"></a>

---

消息入站处理流程

• - Section: 核心模块详解
• - Group: 消息处理与渠道
• - Source File: 15-xiao-xi-ru-zhan-chu-li-liu-cheng.md

本文档系统梳理从用户发送消息到 Agent 生成回复的完整入站链路。覆盖消息接收、安全审计、并发调度、Agent 执行及回复持久化等关键阶段，帮助开发者理解系统如何在多渠道、多消息类型场景下实现有序处理。

概览：三层架构

消息入站处理采用渠道适配层 → 调度层 → 执行层的三层分离架构。渠道适配层负责协议解密和消息格式标准化；调度层管理单会话内的消息并发和队列；执行层完成 Agent 调用与回复持久化。

flowchart TB

    subgraph 渠道适配层

        WX["微信客服<br/>routers/wechat.py"]

        MP["小程序 WebSocket<br/>routers/miniapp/ws.py"]

        CHAT["Chat API<br/>routers/chat.py"]

    end



    subgraph 调度层

        DISPATCH["Session Dispatcher<br/>messaging/session_dispatcher.py"]

        SHORTCUT["快捷方式匹配<br/>messaging/inbound_shortcuts.py"]

        NORM["消息规范化<br/>messaging/inbound.py"]

    end



    subgraph 执行层

        SVC["Agent Service<br/>messaging/agent_service.py"]

        PIPE["图像管道<br/>messaging/image_pipeline.py"]

        AGENT["Agent 运行时<br/>agent.py"]

        DISP_FILT["显示过滤器<br/>messaging/display_filters.py"]

    end



    WX --> DISPATCH

    MP --> DISPATCH

    CHAT --> AGENT



    DISPATCH --> SHORTCUT

    DISPATCH --> NORM

    NORM --> PIPE

    DISPATCH --> SVC

    SVC --> AGENT

    AGENT --> DISP_FILT

三层之间通过回调协议（Protocol） 解耦：渠道层只需实现 TextHandler、ImagePreprocessor、ImageErrorHandler、FollowupArbiter 四个接口，即可接入统一调度。

Sources: session_dispatcher.py, \_\_init\_\_.py

阶段一：渠道接收与解密

每个渠道入口对接各自的传输协议，但最终都标准化为 (content, image_ref) 两种消息类型进入调度层。

微信客服渠道

微信客服消息通过同步拉取模式获取。系统轮询 sync_msg API 拉取新消息批次，对每条消息逐一处理。

接收流程：POST /wechat/callback → XML 解密 → _process_kf_event → async_sync_msg 批量拉取 → _handle_synced_message 逐条处理。

在 _handle_synced_message 中，每条消息依次经历以下关卡：

语音消息在进入文本流程前会先通过阿里云语音转文字服务（_transcribe_wechat_voice_message）转为文本，后续路径与文本消息一致。

Sources: wechat.py, message_access_control.py, message_interception.py

小程序 WebSocket 渠道

小程序通过 WebSocket 长连接接收消息。消息以 JSON 帧传输，无需 XML 解密，但同样需要经过文本审核和访问控制检查。消息通过 dispatch_text 进入统一调度。

Sources: ws.py

Chat API 渠道

/api/chat 端点处理 HTTP 请求，支持流式（SSE）和非流式两种模式。此路径不经过 session_dispatcher 的并发管理，而是直接调用 run_agent_with_memory 或 run_agent_stream_with_memory，但仍需先解析 agent_instruction 和检查 inbound_shortcut。

Sources: chat.py

阶段二：消息规范化

app/messaging/inbound.py 提供渠道无关的消息标准化能力，将不同渠道的原始数据统一为 InboundMessagePart 列表。

flowchart LR

    subgraph 输入格式

        TEXT["纯文本消息"]

        MEDIA["微信 media_id"]

        STORED["已存储 image_id"]

        URL["远程图片 URL"]

    end



    subgraph 规范化

        PART["InboundMessagePart"]

        REF["ImageInputRef"]

    end



    TEXT --> |"from_text()"| PART

    MEDIA --> |"ImageInputRef(kind=wechat_media_id)"| REF

    STORED --> |"ImageInputRef(kind=stored_image_id)"| REF

    URL --> |"ImageInputRef(kind=remote_url)"| REF

    REF --> |"from_image()"| PART

ImageInputRef 统一三种图片来源引用方式，携带 kind 字段区分来源：

prepare_image_message 函数根据 ImageInputRef.kind 分发到不同处理路径：已存储图片直接构建 ImagePipelineResult；其他来源先获取字节流再走图像管道。

Sources: inbound.py

阶段三：快捷方式匹配

在消息进入 Agent 调用之前，系统首先尝试匹配入站快捷方式（inbound_shortcuts）。快捷方式是一组预定义规则，用于对特定关键词返回固定回复，避免触发完整的 Agent 调用链路。

匹配逻辑在两个层级执行：

1. 1. 调度层 — 渠道路由中调用 resolve_inbound_shortcut，匹配后直接发送快捷回复（如微信客服的 wechat_demo_miniprogram 类型）
2. 2. 执行层 — agent_service.py 中再次调用 resolve_inbound_shortcut，匹配后通过 deliver_shortcut_messages 发送富媒体消息（图文卡片、小程序卡片等）

规则定义在 data/i18n/inbound_shortcuts.json 中，支持按渠道过滤（channels）、精确匹配（match_type: exact），以及引用国际化文本（trigger_text_refs）。

Sources: inbound_shortcuts.py

阶段四：会话并发调度

session_dispatcher 是消息入站处理的核心调度器，解决一个关键问题：当用户快速连续发送多条消息时，如何保证 Agent 有序处理而不产生响应冲突。

并发模型

stateDiagram-v2

    [*] --> 空闲

    空闲 --> 文本任务运行 : dispatch_text()

    空闲 --> 图像处理中 : dispatch_image()

    空闲 --> 排队等待 : 队列非空



    文本任务运行 --> 空闲 : 任务完成 → drain_queue()

    文本任务运行 --> 文本任务运行 : 新文本到达 → 取消旧任务/合并内容

    文本任务运行 --> 排队等待 : 新图像到达 → 入队



    图像处理中 --> 排队等待 : 新消息到达 → 入队

    图像处理中 --> 文本任务运行 : 图像完成 → schedule_text_task



    排队等待 --> 文本任务运行 : drain_queue() 弹出文本

    排队等待 --> 图像处理中 : drain_queue() 弹出图像

    排队等待 --> 空闲 : 队列清空

状态管理

调度器使用四个模块级字典跟踪每个会话的状态：

所有状态仅从 asyncio 事件循环单线程修改，无需加锁。

文本消息调度（dispatch_text）

当新文本消息到达时，调度器根据当前状态做出三种决策：

1. 1. 有图像在飞或队列非空 → 消息入队，等待后续 drain
2. 2. 有旧文本任务运行 → 调用 FollowupArbiter 仲裁，支持四种策略：

• - interrupt（默认）：取消旧任务，合并内容为新任务
• - fresh：取消旧任务，以全新内容启动
• - parallel：保留旧任务，新消息入队
• - handled：旧任务继续，新消息已被拦截处理

1. 3. 无任务运行 → 直接调度新任务

新任务创建后会进入 0.3 秒防抖窗口（debounce），在此期间如果被更新的任务取消，就不会调用 Agent。防抖通过 dispatch_guard 中的 ContextVar 实现任务身份验证，确保过期任务不会误执行。

Sources: session_dispatcher.py, dispatch_guard.py

图像消息调度（dispatch_image）

图像消息具有最高优先级：

1. 1. 如果有正在运行的文本任务 → 取消它，保留其内容加入图像批次
2. 2. 标记图像处理开始（mark_image_in_flight）
3. 3. 启动后台任务 _run_accumulated_image_batch

图像批处理循环会持续从队列中弹出消息，预处理所有图像，最后将合并的文本+图像结果交给文本任务路径。如果图像管道失败（如内容审核不通过），会通过 on_image_error 回调通知渠道层。

Sources: session_dispatcher.py

阶段五：图像管道

图像管道（image_pipeline）处理图像从接收到可被 Agent 使用的完整生命周期：

flowchart LR

    A["原始图像字节"] --> B["内容审核<br/>ensure_image_content_allowed"]

    B --> C["上传 OSS + 去重检测<br/>ingest_user_uploaded_image"]

    C --> D{去重?}

    D -->|是| E["返回已有 image_id<br/>is_duplicate=True"]

    D -->|否| F["保存图像 + 图像描述生成"]

    F --> G["构建 agent_message"]

    G --> H["ImagePipelineResult"]

    E --> H

去重阈值默认为 0.97（余弦相似度），超过此阈值的图像被视为重复，跳过 Agent 调用。去重检测通过 ingest_user_uploaded_image 在数据库级别完成。

ImagePipelineResult 包含三个关键字段：

• - image_id — 数据库中的图像 ID
• - agent_message — 面向 Agent 的描述文本，格式如 [用户发送了一张图片，图片ID=123]
• - user_message — 面向持久化的结构化 JSON 载荷

Sources: image_pipeline.py, agent_messages.py

阶段六：Agent 执行（run_agent_turn）

agent_service.py 中的 run_agent_turn 函数是 Agent 调用的核心编排器，负责消息持久化、Agent 运行、回复发送的全流程管理。

完整执行流程

flowchart TB

    START["run_agent_turn"] --> CONV["获取或创建会话<br/>get_or_create_conversation"]

    CONV --> PERSIST_USER["持久化用户消息<br/>save_user_message"]

    PERSIST_USER --> REG_REPLY["注册回复状态<br/>register_pending_reply_keys"]

    REG_REPLY --> BUILD_CTX["构建 ToolContext"]

    BUILD_CTX --> LOAD_IMG["加载模型图像 + 历史图像"]

    LOAD_IMG --> SHORTCUT_CHECK{快捷方式<br/>匹配?}



    SHORTCUT_CHECK -->|是| DELIVER_SHORTCUT["deliver_shortcut_messages"]

    SHORTCUT_CHECK -->|否| STALE_CHECK_1{是否过期?}



    DELIVER_SHORTCUT --> STALE_CHECK_2{是否过期?}

    STALE_CHECK_1 -->|过期| DROP_1["丢弃结果"]

    STALE_CHECK_1 -->|有效| RUN_AGENT["run_agent_with_memory"]

    STALE_CHECK_2 -->|过期| DROP_2["丢弃结果"]

    STALE_CHECK_2 -->|有效| SEND_REPLY_CHECK



    RUN_AGENT --> STALE_CHECK_3{是否过期?}

    STALE_CHECK_3 -->|过期| DROP_3["丢弃结果"]

    STALE_CHECK_3 -->|有效| SEND_REPLY_CHECK{send_reply<br/>回调?}



    SEND_REPLY_CHECK -->|有| SEND["发送回复到渠道"]

    SEND_REPLY_CHECK -->|无| PERSIST_REPLY["持久化助手消息"]

    SEND --> PERSIST_REPLY

    PERSIST_REPLY --> MARK_SENT["标记回复已发送<br/>mark_session_reply_sent"]

    MARK_SENT --> RESULT["返回 AgentTurnResult"]

过期任务检测

在 Agent 执行过程中，每个关键步骤之间都会调用 is_current_dispatch_task(session_id) 检查当前任务是否仍是最新的。如果在 Agent 运行期间用户发送了新消息，旧任务会被取消或被更新的任务取代，此时旧任务的输出会被静默丢弃，避免向用户发送过时回复。

回复状态跟踪

reply_state 模块（messaging/reply_state.py）通过 Redis 跟踪每条消息的回复状态，防止重复回复。关键机制：

• - register_latest_user_message — 记录最新用户消息 ID
• - register_pending_reply_keys — 注册待回复的消息标识
• - mark_session_reply_sent — 标记回复已完成（支持 phase 阶段标记）

Sources: agent_service.py, reply_state.py

阶段七：显示过滤与回复发送

Agent 生成的回复在发送给用户前会经过显示过滤器（display_filters.py）处理，移除内部推理标记（如 </nuva_thought> 之前的内容）。这确保用户只看到最终结论，而非 Agent 的中间思考过程。

过滤规则为硬编码的元组：

• - strip_before_including — 移除指定标记及其之前的所有内容
• - remove_between_including — 移除两个标记之间的内容（含标记）

渠道层在 send_reply 回调中应用过滤后再发送。以微信客服为例，过滤后的文本还会经过 rewrite_wechat_mp_links_for_chat 将小程序链接转为微信可解析格式，再通过 build_wechat_reply_delivery 决定是以纯文本还是菜单消息形式发送。

Sources: display_filters.py, wechat.py

后续消息仲裁（Followup Arbiter）

当用户在 Agent 处理过程中连续发送消息时，FollowupArbiter 决定如何处理新消息。微信渠道实现了自定义仲裁器 arbitrate_wechat_followup（wechat_followup_intent.py），它结合 Agno 运行时快照来判断：

• - Agent 是否有活跃的远程任务（如图像生成工作流）
• - 是否应中断当前任务、合并内容、还是并行排队

仲裁结果为 FollowupDecision，包含 action（interrupt/parallel/handled/fresh）、可选的 replacement_content 和 message_records。

Sources: wechat_followup_intent.py, session_dispatcher.py

关键设计模式总结

相关文档

• - 微信渠道的详细集成细节参见 微信客服与小程序集成
• - 会话与回复状态的深入设计参见 会话分发与回复状态
• - Agent 架构和技能系统参见 单主 Agent 架构
• - Dialog State 的设计参见 Dialog State 设计
• - 工具系统的注册与工厂参见 工具注册与工厂

<a id="16-wei-xin-ke-fu-yu-xiao-cheng-xu-ji-cheng"></a>

---

微信客服与小程序集成

• - Section: 核心模块详解
• - Group: 消息处理与渠道
• - Source File: 16-wei-xin-ke-fu-yu-xiao-cheng-xu-ji-cheng.md

本页文档详解系统与微信生态的两条集成通道——企业微信客服（KEFU）和微信小程序（Mini Program）。二者共享统一的 Agent 核心和消息调度框架，但在入站协议、出站投递和用户身份体系上各具特点。理解两条通道的差异与共性，是进行渠道定制或排查跨渠道问题的前提。

架构总览

系统通过两条并行的微信通道与用户交互，它们在入口层分流，但汇聚到同一套消息调度器和 Agent 运行时。

flowchart TB

    subgraph WeChat["微信生态"]

        UserK["企业微信客服用户"]

        UserM["小程序用户"]

    end



    subgraph Entry["入口层"]

        WB["POST /wechat/callback<br/>XML Webhook"]

        MC["POST /api/wechat/chat<br/>HTTP REST"]

        MW["WS /api/wechat/ws<br/>WebSocket"]

    end



    subgraph WechatModule["app/wechat/ 模块"]

        Crypto["crypto.py<br/>AES 加解密"]

        API["api.py<br/>企业微信 API 封装"]

        MenuMD["menu_markdown.py<br/>菜单消息渲染"]

        MiniprogramRoutes["miniprogram_routes.py<br/>小程序路由解析"]

    end



    subgraph Messaging["app/messaging/ 统一调度层"]

        Dispatcher["session_dispatcher.py<br/>会话消息调度器"]

        Inbound["inbound.py<br/>入站消息归一化"]

        AgentSvc["agent_service.py<br/>Agent 执行服务"]

        MiniappWS["miniapp_ws.py<br/>WebSocket 连接注册"]

        ReplyState["reply_state.py<br/>回复状态跟踪"]

    end



    subgraph Routers["路由层"]

        WechatRouter["routers/wechat.py<br/>客服回调路由"]

        MiniappChatRouter["routers/miniapp/chat.py<br/>小程序聊天路由"]

        MiniappWsRouter["routers/miniapp/ws.py<br/>小程序 WS 路由"]

    end



    UserK -->|"XML/加密"| WB

    UserM -->|"JSON/Token"| MC

    UserM -->|"WS 连接"| MW



    WB --> WechatRouter

    MC --> MiniappChatRouter

    MW --> MiniappWsRouter



    WechatRouter --> Crypto

    WechatRouter --> API

    WechatRouter --> Dispatcher

    MiniappChatRouter --> AgentSvc

    MiniappWsRouter --> MiniappWS

    MiniappWsRouter --> AgentSvc



    API -->|"send_msg"| WeChat

    MiniappWS -->|"推送事件"| UserM



    Dispatcher --> Inbound

    Dispatcher --> AgentSvc

两条通道的核心差异如下表所示：

Sources: config.py, main.py

企业微信客服通道

企业微信客服通道通过 Webhook 回调接收消息，经 AES 解密后进入统一调度流程。系统以"拉取模式"同步客服消息——收到事件通知后调用 kf/sync_msg 拉取实际消息体。

Webhook 回调处理

企业微信客服消息的入口是 POST /wechat/callback。该端点负责两件事：URL 验证（GET 请求）和消息事件接收（POST 请求）。事件到达后经过签名验证、AES 解密和 XML 解析，仅处理 kf_msg_or_event 类型的事件。

消息处理流程如下：

sequenceDiagram

    participant WX as 企业微信服务器

    participant Router as /wechat/callback

    participant Sync as sync_msg 拉取

    participant Handler as _handle_synced_message

    participant Dispatcher as session_dispatcher

    participant Agent as Agent 运行时

    participant API as kf/send_msg



    WX->>Router: POST XML（加密事件）

    Router->>Router: AES 解密 + 签名验证

    Router->>Sync: async_sync_msg(cursor)

    Sync-->>Router: msg_list + next_cursor



    loop 每条消息

        Router->>Handler: _handle_synced_message(msg)

        Handler->>Handler: 去重（Redis msgid）

        Handler->>Handler: ensure_user_oauth（创建/关联用户）

        Handler->>Handler: 内容审核 + 权限校验

        Handler->>Dispatcher: dispatch_text / dispatch_image

        Dispatcher->>Agent: run_agent_with_memory

        Agent-->>Dispatcher: reply

        Dispatcher->>API: send_text_msg / send_menu_msg

    end



    Router-->>WX: "success"

关键设计点在于消息去重和用户身份映射。系统使用 Redis 以 wechat:kefu:msg:{msgid} 为键做 24 小时去重窗口；同时通过 ensure_user_oauth 将企业微信的 external_userid 映射为数据库用户 ID，支持 unionid 跨渠道关联。

Sources: routers/wechat.py, routers/wechat.py, wechat/api.py

消息加解密

企业微信使用 AES-CBC 模式对回调消息进行加密。WXBizMsgCrypt 类封装了加解密逻辑，使用 encoding_aes_key 生成密钥，corp_id 作为身份标识参与校验。签名验证采用 SHA1 哈希，将 token、timestamp、nonce、encrypt 排序后拼接比较。

Sources: wechat/crypto.py

用户欢迎语与小程序卡片

首次进入客服会话的用户会触发 welcome_code 事件。系统通过 send_msg_on_event API 发送欢迎消息，支持纯文本和 msgmenu（菜单消息）两种格式。菜单消息中可以嵌入小程序卡片，引导用户从客服场景跳转到小程序获得完整功能体验。

欢迎语的格式由 wechat_welcome 配置项控制，支持三种菜单项类型：

小程序卡片发送需要 thumb_media_id（缩略图素材 ID）。系统支持自动上传和缓存缩略图——当配置了 WECHAT_MINIPROGRAM_IMAGE_URL 时，会自动上传图片并缓存 media_id 到 Redis（TTL 2.5 天），过期后自动刷新。

Sources: routers/wechat.py, routers/wechat.py, wechat/api.py

消息发送能力

企业微信客服 API（kf/send_msg）支持多种消息类型，系统通过 app/wechat/api.py 封装了同步和异步两套接口：

所有 API 调用都经过统一的 _call_wechat_api_async 包装，内置性能日志（elapsed_ms）和错误码解析。access_token 通过 Redis 缓存（TTL 7000 秒），避免重复请求。

系统还支持将 Agent 输出中的 Markdown 菜单语法自动转换为微信 msgmenu 格式。menu_markdown.py 负责解析 选项文本 语法，生成包含 click、view、miniprogram 等菜单项的消息体。

Sources: wechat/api.py, wechat/api.py, wechat/menu_markdown.py

语音消息处理

当用户发送语音消息时，系统下载原始音频（AMR 格式），调用阿里云语音识别服务（transcribe_audio_bytes）转写为文本，然后作为普通文本消息进入调度流程。转写失败时返回错误提示。

Sources: routers/wechat.py, routers/wechat.py

图片处理与提示

当用户发送图片时，系统会启动延迟提示机制。在图片开始处理后 3 秒，如果处理尚未完成，系统会发送一条随机的"正在处理"提示消息，提升用户等待体验。提示文本按图片数量分桶（single/few/batch）并支持 visual 和 fun 两种风格，按 7:3 权重随机选择。

图片消息进入 dispatch_image 调度后，系统会：

1. 1. 注册图片处理中的状态（mark_image_in_flight）
2. 2. 下载微信图片素材（kf/get_media）
3. 3. 运行图片处理流水线
4. 4. 将图片描述注入 Agent 上下文
5. 5. 完成后清除处理状态

Sources: routers/wechat.py, routers/wechat.py

微信小程序通道

小程序通道提供了比客服通道更丰富的交互能力。它包含 HTTP REST API 和 WebSocket 两种接入方式，支持流式对话和实时事件推送。

HTTP API 接口

小程序 HTTP API 以 /api/wechat/ 为前缀，需要 JWT Token 认证（通过 LoginRequired 依赖注入）。主要端点包括：

聊天端点 POST /api/wechat/chat 接受 MiniappChatRequest，支持两种消息格式：

• - legacy 格式：message 字段传递纯文本
• - multipart 格式：messages 数组支持 text 和 image 混合输入

请求处理流程：构建入站消息部件 → 内容审核 → 访问控制 → Agent 执行 → 返回回复。

Sources: routers/miniapp/chat.py, routers/miniapp/chat.py

WebSocket 实时通道

WebSocket 端点 WS /api/wechat/ws 提供双向实时通信能力。连接建立时通过 resolve_ws_user 验证 Token（支持 query param 和 Authorization header 两种方式），注册到连接池 miniapp_ws.py。

WebSocket 通道的核心能力包括：

1. 1. 流式对话：消息通过 WebSocket 发送，Agent 回复逐 token 推送
2. 2. 穿搭结果推送：后台生成完成后，通过 send_miniapp_event 实时推送穿搭卡片
3. 3. 状态同步：支持 outfit 轮询模式——客户端发送 outfit 请求后，服务端以 0.8 秒间隔轮询数据库状态，完成后立即推送

连接管理通过 _user_connections 字典维护（user_id → set[WebSocket]），支持同一用户多个连接。发送失败的连接自动清理。

flowchart LR

    subgraph Client["小程序客户端"]

        ChatUI["聊天界面"]

    end



    subgraph Server["服务端"]

        WS["WS 端点"]

        ConnPool["连接池<br/>miniapp_ws"]

        Agent["Agent 运行时"]

        OutfitPoll["Outfit 轮询器"]

    end



    ChatUI <-->|"text / image"| WS

    WS <--> ConnPool

    WS --> Agent

    Agent -->|"reply chunk"| WS

    OutfitPoll -->|"outfit card push"| ConnPool

    ConnPool -->|"event"| ChatUI

Sources: routers/miniapp/ws.py, messaging/miniapp_ws.py, routers/miniapp/ws_common.py

小程序登录与用户身份

小程序用户通过 code 换取 JWT Token 完成登录。user.py 中的 /api/wechat/user/login 端点接收微信 login 接口返回的 code，调用微信 jscode2session API 获取 openid 和 session_key，然后生成应用层 JWT Token。

Token 验证通过 get_token_data 函数完成，支持开发环境的 DEV_TOKEN 直通模式。用户身份最终映射为数据库中的 user_id（整数主键），与客服通道的 external_userid 通过 unionid 关联。

Sources: routers/miniapp/user.py, routers/miniapp/ws_common.py

统一消息调度框架

两条通道的差异在入口层处理后，汇聚到 app/messaging/ 下的统一调度框架。这一框架的核心是 session_dispatcher.py 中的会话状态机，它保证每个会话同时只有一个 Agent 调用在运行。

会话调度器

调度器管理三类并发状态：

调度规则如下：

• - 图片阻塞：图片处理期间，后续所有消息排队等待
• - 文本合并：新文本到达时，可取消正在等待的旧文本任务并合并内容（防抖行为）
• - 自动排空：任何任务完成后自动排空队列

渠道适配器只需提供两个回调：on_text（处理文本消息）和 prepare_image（解析图片引用）。客服通道通过 _make_wechat_handlers 生成绑定 open_kfid 和 external_userid 的闭包；小程序通道直接在路由中调用 Agent 服务。

Sources: messaging/session_dispatcher.py, messaging/session_dispatcher.py

入站消息归一化

inbound.py 将不同渠道的入站消息统一为 InboundMessagePart 列表，支持 text 和 image 两种部件类型。图片引用通过 ImageInputRef 数据类标准化，支持三种来源：

Sources: messaging/inbound.py, messaging/inbound.py

Agent 执行服务

agent_service.py 提供 run_inbound_agent_turn 函数，封装了完整的 Agent 执行流程：创建会话 → 保存用户消息 → 执行 Agent → 保存助手回复 → 标记回复状态。两条通道都通过此函数驱动 Agent，确保消息持久化和状态管理的一致性。

Sources: messaging/agent_service.py, messaging/agent_service.py

渠道专属配置

系统通过 data/channels/ 目录为不同渠道提供专属的提示词和技能配置。每个渠道目录下可包含：

当前支持的渠道目录：

• - wechat_kefu — 企业微信客服
• - wechat_miniapp — 微信小程序

channel_prompts.py 提供了 load_optional_channel_prompt 函数，按 channels/{channel_name}/{relative_path} 路径加载渠道专属内容。如果文件不存在则返回空字符串，实现优雅降级。

Sources: channel_prompts.py

环境变量配置

微信集成相关的环境变量如下：

客服通道要求四个 WECHAT_* 变量全部配置，否则返回 503。小程序通道仅需 JWT 认证，不依赖客服相关配置。

Sources: .env.example, config.py

小程序路由映射

miniprogram_routes.py 定义了系统支持的小程序页面路由映射，用于将 Agent 输出中的 mp:// 协议链接转换为小程序原生页面跳转：

在客服通道中，这些链接被转换为 Markdown 菜单项；在小程序通道中，被转换为纯文本或 WebSocket 选项。

Sources: wechat/miniprogram_routes.py, wechat/miniprogram_routes.py

内容安全审核

两条通道的消息都经过阿里云内容安全审核，但策略有所不同：

config.yml 中 aliyun_content_audit_enabled_channels 默认仅包含 wechat_miniapp，即客服通道默认跳过内容审核。这是基于客服通道误报率较高的生产实践调优。

Sources: config.yml, routers/wechat.py

消息准入控制

系统提供统一的消息准入控制机制，通过 message_access_control.py 实现。准入检查在两条通道的消息处理入口处执行，支持白名单、黑名单和授权码三种模式。

准入检查的参数包括渠道标识、服务账号（客服通道为 open_kfid）、用户身份（user_id + openid + unionid）。被拒绝的用户会收到国际化错误提示。

Sources: routers/wechat.py, routers/miniapp/chat.py

调试与运维

客服通道关键日志标签：

• - [wechat_api:start/done/error] — 企业微信 API 调用耗时与结果
• - [wechat:sync] — 消息同步状态
• - [wechat:greeting] — 欢迎语发送

小程序通道关键日志标签：

• - [miniapp-ws] — WebSocket 连接生命周期
• - [miniapp-chat] — HTTP 聊天请求

Redis 键清单（客服通道）：

<a id="17-hui-hua-fen-fa-yu-hui-fu-zhuang-tai"></a>

---

会话分发与回复状态

• - Section: 核心模块详解
• - Group: 消息处理与渠道
• - Source File: 17-hui-hua-fen-fa-yu-hui-fu-zhuang-tai.md

当用户在短时间内连续发送多条消息（文本+图片、多条文本等），系统必须保证 Agent 按顺序处理且不产生交错回复。本页详细解析 会话分发器（Session Dispatcher） 如何管理并发消息的调度，以及 回复状态（Reply State） 如何在 Redis 中追踪每条消息的回复进度，防止重复回复和状态漂移。

分发状态机概览

整个会话分发机制以模块级字典维护状态，不依赖数据库锁，仅在 asyncio 事件循环中被安全修改。下图展示了一条消息从入站到 Agent 处理完成的完整生命周期：

stateDiagram-v2

    [*] --> Incoming: 用户发送消息

    Incoming --> CheckQueue: dispatch_text / dispatch_image



    state CheckQueue {

        direction LR

        [*] --> ImageInFlight: 图片正在处理?

        [*] --> TextRunning: 文本任务正在运行?

        [*] --> QueueNotEmpty: 队列非空?

        [*] --> Ready: 均否

    }



    ImageInFlight --> Enqueued: 入队等待

    TextRunning --> ArbiterCheck: 是否有仲裁器?

    QueueNotEmpty --> Enqueued: 入队等待



    ArbiterCheck --> MergeAndCancel: interrupt / fresh

    ArbiterCheck --> ParallelEnqueue: parallel

    ArbiterCheck --> DropMessage: handled



    Ready --> ScheduleTask: 调度文本任务

    MergeAndCancel --> ScheduleTask: 合并内容后调度



    ScheduleTask --> DebounceSleep: 300ms 去抖

    DebounceSleep --> ExecuteAgent: 调用 on_text

    ExecuteAgent --> DrainQueue: 完成后自动排空队列

    DrainQueue --> [*]: 下一条消息

Sources: session_dispatcher.py

核心状态字典

会话分发器使用四个模块级字典追踪每个 session_id 的并发状态。这些字典仅在 asyncio 事件循环中被修改，无需锁机制——前提是部署模型为单事件循环的 FastAPI 进程。

当一个 Agent 任务完成后，调度器在 finally 块中自动调用 drain_queue 来处理队列中下一条消息，形成链式排空。

Sources: session_dispatcher.py

消息分发入口函数

调度器对外暴露三个入口函数，渠道路由（如微信客服、小程序）通过它们驱动状态机，从不直接操作内部字典。

dispatch_text

文本消息的分发入口。其核心逻辑分为三条路径：

1. 1. 阻塞路径：若图片正在处理或队列非空，消息直接入队等待
2. 2. 仲裁路径：若有正在运行的文本任务，通过 FollowupArbiter 决策后行动
3. 3. 直接调度路径：无阻塞、无运行任务时，立即创建新的文本任务

flowchart TD

    A[dispatch_text] --> B{图片处理中<br>或队列非空?}

    B -->|是| C[enqueue → 返回]

    B -->|否| D{有运行中的<br>文本任务?}

    D -->|否| H[_schedule_text_task]

    D -->|是| E[调用 arbiter]

    E --> F{arbiter 决策}

    F -->|interrupt| G[合并内容,<br>取消旧任务]

    F -->|fresh| G2[丢弃旧内容,<br>取消旧任务]

    F -->|parallel| I[消息入队,<br>不中断旧任务]

    F -->|handled| J[旧任务保持,<br>新消息丢弃]

    G --> H

    G2 --> H

关键设计：文本任务创建后会先经历 300ms 去抖睡眠（asyncio.sleep(0.3)），在此窗口内到达的新消息可通过仲裁器合并或取消，避免用户快速输入时产生多次 Agent 调用。

Sources: session_dispatcher.py

dispatch_image

图片消息的分发入口。图片具有最高优先级——若当前有文本任务正在运行，图片到达后会取消该文本任务（保留其内容用于后续合并），然后立即启动图片处理流程。

图片处理完成后，调度器会将累积的图片描述和后续文本合并为一条消息，交给正常的可取消文本任务路径处理。这意味着图片处理期间用户发送的文本不会丢失。

Sources: session_dispatcher.py

drain_queue

队列排空函数是整个调度器的"心脏"。它在每次任务完成后被自动调用，从数据库中弹出待处理消息并按类型分发：

• - 若弹出消息中包含图片：标记图片处理中，启动 _run_accumulated_image_batch 后台任务
• - 若弹出消息全是文本：合并多条文本内容后调度新的文本任务
• - 若队列为空或有活跃任务：终止排空循环

重入保护机制通过 _queue_draining 集合实现——同一个 session 的重复排空调用会被静默丢弃。

Sources: session_dispatcher.py

后续消息仲裁（Followup Arbiter）

当用户在 Agent 任务运行期间发送新消息时，调度器需要决定如何处理。系统通过 FollowupArbiter 协议支持可插拔的决策策略，微信客服渠道使用基于 LLM 的意图仲裁器来自动分类：

仲裁器构建上下文快照时会并发加载近期对话历史、用户记忆和用户画像，为 LLM 决策提供充足信息。

Sources: wechat_intent_arbitration.py, session_dispatcher.py

回复状态追踪

回复状态模块在 Redis 中维护每条用户消息的回复进度，解决两个核心问题：快速回复（Fast Response）的准确性和消息级回复去重。

Redis 键结构

回复状态使用五类 Redis 键，统一使用 7 天 TTL：

Sources: reply_state.py

回复状态生命周期

一条用户消息从到达到回复完成的状态变化流程如下：

sequenceDiagram

    participant Channel as 渠道路由

    participant RS as Reply State (Redis)

    participant Dispatcher as Session Dispatcher

    participant Agent as Agent Service



    Channel->>RS: register_pending_reply_keys(session, keys)

    Channel->>RS: register_latest_user_message(session, msg_id)

    Note over RS: counter:{msg_id}.reply_count = 0



    Dispatcher->>Agent: on_text(session, content, ...)

    Agent->>Agent: run_agent_with_memory(...)



    alt 快速回复可用

        Agent->>RS: can_send_fast_response(session, msg_id)

        RS-->>Agent: true (reply_count == 0)

        Agent->>Channel: 发送快速回复

    end



    Agent->>Channel: 发送最终回复

    Agent->>RS: mark_session_reply_sent(session, phase="final")

    Note over RS: counter:{msg_id}.reply_count += 1<br>pending keys 清空<br>message:{key}.replied = true

Sources: reply_state.py, agent_service.py

快速回复守卫

can_send_fast_response 函数通过比较消息 ID 和回复计数来判断是否允许发送快速回复。条件为：消息 ID 匹配且 reply_count 为 0。一旦 mark_session_reply_sent 被调用（回复计数 +1），快速回复就被自动禁用——这确保了快速回复只在 Agent 尚未输出最终回复时生效。

Sources: reply_state.py, fast_response.py

回复种子机制

register_pending_latest_reply_seed 提供了一种跨消息传递回复计数的机制。当渠道（如微信客服）在用户消息到达前就已发送了一条通知回复（如"图片处理中…"），它可以通过 seed 预先累加回复计数。当 register_latest_user_message 被调用时，seed 值会被继承到新消息的 counter 中，确保 can_send_fast_response 对该消息返回 false。

Sources: reply_state.py

分发任务守卫（Dispatch Guard）

dispatch_guard 模块使用 Python 的 ContextVar 实现协程级别的任务身份验证，防止被取消的过期任务向用户发送回复。

每当新文本任务被创建时，调度器会生成一个唯一 token 并绑定到 ContextVar。任务内部的关键节点（如 Agent 调用前后）通过 is_current_dispatch_task 检查自身是否仍为当前活跃任务。若 token 已被更新（说明新消息已到达并替换了该任务），过期任务将静默中止，不会执行后续回复。

flowchart LR

    A[新消息到达] --> B[生成 task_token = uuid4]

    B --> C[set_current_dispatch_token]

    C --> D[创建 asyncio.Task]

    D --> E{is_current_dispatch_task?}

    E -->|是| F[继续执行 on_text]

    E -->|否| G[静默中止]

    F --> H[发送回复]

    H --> I[clear_current_dispatch_token]

Sources: dispatch_guard.py, session_dispatcher.py

渠道集成模式

各渠道路由通过实现回调协议来接入分发器。以微信客服为例，它需要提供四个回调：

入站消息经过访问控制、内容审核、快捷方式匹配等前置处理后，才进入 dispatch_text / dispatch_image。消息的 reply_state_key（如微信 msgid）在入口处即被注册到 Redis，确保后续的回复状态追踪覆盖完整生命周期。

Sources: wechat.py, wechat.py

图片批处理流程

当图片到达时，调度器不会立即调用 Agent——而是启动 _run_accumulated_image_batch 后台任务。该任务会持续排空队列，预处理所有排队的图片（上传 OSS、去重检测、生成描述），然后将合并后的图片描述 + 累积文本一起交给可取消的文本任务路径。

这种设计确保了：用户连续发送 5 张图片时，系统会等待所有图片处理完成后再调用 Agent，而非逐张处理产生 5 次独立对话。

Sources: session_dispatcher.py

并发安全总结

整个会话分发与回复状态系统通过三层机制保证并发安全：

这三层机制共同保证了：即使在高频消息场景下（用户快速连发、图片+文本混合），系统也绝不会出现交错回复或重复回复。

<a id="18-mem0-ji-yi-guan-li"></a>

---

Mem0 记忆管理

• - Section: 核心模块详解
• - Group: 记忆与洞察
• - Source File: 18-mem0-ji-yi-guan-li.md

本页介绍 outfit_agent 的长期用户记忆系统——基于 Mem0 框架构建的语义记忆层。该系统使 Agent 能够跨会话积累用户偏好（体型特征、风格倾向、预算区间、场景禁忌等），并在每次对话前自动注入相关记忆上下文，从而实现真正的"越用越懂你"个性化穿搭服务。

记忆系统由两条并行链路驱动：在线实时提取（每轮对话后可选执行）和离线每日洞察分析（定时批任务，基于穿搭记录、衣橱单品与聊天历史的多维度深度分析）。两条链路最终都将结构化事实写入同一个 Mem0 存储，由 Agent 在对话发起时统一检索。

系统架构总览

Mem0 记忆系统的核心设计可概括为 "先读后写、双链并行、单一存储"。下图展示了从用户消息到达至记忆写入的完整数据流：

flowchart TB

    subgraph Online["在线链路（每轮对话）"]

        U[用户消息] --> S[search_user_memory<br/>语义检索相关记忆]

        S --> |格式化记忆上下文| A[Agent 推理]

        A --> R[生成回复]

        R --> |开启时| W[add_conversation_memory<br/>实时提取事实]

    end



    subgraph Offline["离线链路（每日定时）"]

        CRON[每日 00:00 调度] --> SEL[InsightSelector<br/>扫描候选用户]

        SEL --> COL[InsightCollector<br/>收集三维增量数据]

        COL --> ANA[InsightAnalyzers<br/>分维度 LLM 分析]

        ANA --> MERGE[analyze_merge<br/>冲突仲裁引擎]

        MERGE --> SYNC[Mem0Synchronizer<br/>执行 add/update/delete]

    end



    subgraph Storage["统一存储"]

        W --> PG[(pgvector<br/>outfit_mem0_memories)]

        SYNC --> PG

        S --> PG

    end



    subgraph Admin["管理接口"]

        PG --> API[Admin API<br/>查看/编辑/删除记忆]

        PG --> DASH[Dashboard<br/>记忆浏览面板]

    end



    style Online fill:#e8f5e9,stroke:#2e7d32

    style Offline fill:#e3f2fd,stroke:#1565c0

    style Storage fill:#fff3e0,stroke:#e65100

    style Admin fill:#f3e5f5,stroke:#6a1b9a

两条链路的核心差异：

Sources: app/memory/__init__.py, app/memory/ops.py, app/insights/service.py

初始化与配置

启动流程

Mem0 以 AsyncMemory 单例 模式运行，在 FastAPI 应用启动阶段通过 init_memory() 一次性初始化。该初始化嵌入在 AgentOS 的生命周期管理中，由 app/main.py 的 _mem0_lifespan 控制：

sequenceDiagram

    participant Main as app/main.py

    participant Lifespan as _mem0_lifespan

    participant Client as memory/client.py

    participant Mem0 as AsyncMemory

    participant PG as pgvector (PostgreSQL)



    Main->>Lifespan: FastAPI lifespan 启动

    Lifespan->>Lifespan: DB schema 迁移

    Lifespan->>Client: init_memory()

    Client->>Client: 解析 database_url → pgvector 参数

    Client->>Client: 构建 MemoryConfig

    Client->>Mem0: AsyncMemory(config)

    Mem0->>PG: 连接并验证 collection

    Client-->>Lifespan: 单例就绪

    Lifespan-->>Main: 启动完成

init_memory() 内部完成三项关键配置组装：

1. 1. 向量存储（pgvector）：使用项目现有的 PostgreSQL 数据库，collection 名称默认为 outfit_mem0_memories，embedding 维度与 text_embedding 模型配置一致（默认 1024）。
2. 2. LLM 提取器：使用 flash 模型（默认 temperature=0.1）驱动记忆事实提取。
3. 3. Embedding 模型：使用 text_embedding 模型生成语义向量，通过 OpenAI 兼容接口调用。

Sources: app/memory/client.py, app/main.py

配置参数

所有记忆相关配置通过环境变量或 config.yml 控制，由 app/config.py 的 Settings 类统一管理：

离线洞察分析的额外配置：

Sources: app/config.py, .env.example

在线记忆操作

在线链路是 Agent 每次响应用户消息时的记忆读写路径，由 app/memory/ops.py 封装为三个核心操作。

记忆检索（Read-Before）

search_user_memory() 是 Agent 运行前的记忆注入入口。它执行 双重语义搜索：先搜全局用户画像记忆（agent_id 维度），再搜当前会话上下文记忆（run_id 维度），合并去重后格式化为结构化文本块注入 Agent 输入。

flowchart LR

    Q[用户消息] --> Search[search_user_memory]

    Search --> |query + user_id| P[Profile 记忆<br/>agent_id 维度]

    Search --> |query + user_id + session_id| S[Session 记忆<br/>run_id 维度]

    P --> Merge[合并去重<br/>按 memory id]

    S --> Merge

    Merge --> Format[格式化输出]

    Format --> |---用户记忆---<br/>[用户画像]<br/>- 记忆1<br/>- 记忆2<br/>[本次会话上下文]<br/>- 记忆3<br/>--------------| Inject[注入 agent_input]

格式化后的记忆文本按 [用户画像] 和 [本次会话上下文] 两个区块组织，直接拼接到用户消息前面，使 Agent 在推理时能自动参考历史偏好。

在 app/agent.py 的 run_agent_with_memory() 中，检索延迟和记忆字符数会被记录用于可观测性监控：

[agent:memory] user=<id> session=<id> elapsed_ms=<ms> memory_chars=<len>

Sources: app/memory/ops.py, app/agent.py

记忆写入（Write-After）

add_conversation_memory() 在 Agent 完成回复后异步执行。它将当前轮的 user+assistant 消息对提交给 Mem0，由 Mem0 内部的 LLM 根据 mem_extraction.zh.txt 中定义的提取规则自动识别并存储事实。

写入操作通过 asyncio.Semaphore(2) 限制并发数，避免 Mem0 内部同步 psycopg2 连接池被过多异步任务压垮。所有写入异常均被捕获并记录为 warning，确保记忆故障不会阻断主响应路径。

实时提取默认关闭（mem0_realtime_analysis_enabled=false），推荐依赖离线洞察获得更全面、经过冲突仲裁的记忆质量。

提取提示词定义了 5 大提取维度：

Sources: app/memory/ops.py, data/prompts/memory/mem_extraction.zh.txt, app/agent.py

用户风格画像查询

get_user_style_profile() 提供面向 Agent 工具调用的风格画像快捷接口。它使用固定的风格相关查询词进行语义搜索，只返回与穿搭决策相关的记忆子集，而非全量记忆：

"用户风格偏好 体型特征 预算区间 着装禁忌 品牌偏好"

该函数与 get_user_profile()（返回全量记忆列表）形成互补——前者用于 Agent prompt 注入，后者用于管理面板展示。

Sources: app/memory/ops.py, app/memory/ops.py

离线洞察分析链路

离线链路是记忆系统的核心深度引擎，由 app/insights/ 模块实现。它通过每日定时任务扫描活跃用户，从穿搭记录、衣橱单品和聊天历史三个维度提取深层洞察，经冲突仲裁后同步到 Mem0。

整体流程

flowchart TB

    START[每日 00:00<br/>scheduler 触发] --> SCAN[InsightSelector<br/>扫描全部用户]

    SCAN --> TRIGGER{是否满足<br/>触发条件?}

    TRIGGER --> |否| SKIP[跳过]

    TRIGGER --> |是| COLLECT[InsightCollector<br/>收集增量数据]



    COLLECT --> O{新增穿搭<br/>>= 阈值?}

    COLLECT --> I{新增单品<br/>>= 阈值?}

    COLLECT --> C{距上次聊天洞察<br/>>= 30天?}



    O --> |是| AO[analyze_outfit<br/>穿搭维度分析]

    I --> |是| AI[analyze_item<br/>衣橱维度分析]

    C --> |是| AC[analyze_chat<br/>对话维度分析]

    C --> |是| AT[analyze_tone<br/>语调风格分析]



    AO --> SAVE1[写入 user_insights]

    AI --> SAVE1

    AC --> SAVE1

    AT --> SAVE1



    SAVE1 --> AGG[聚合数据]

    AGG --> |穿搭: 近2年洞察<br/>衣橱: 最新一次<br/>对话: 近2年洞察<br/>+ 当前 Mem0 记忆| MERGE[analyze_merge<br/>冲突仲裁引擎]

    MERGE --> SYNC[Mem0Synchronizer<br/>apply_actions]

    SYNC --> |add/update/delete| MEM0[(Mem0<br/>pgvector)]



    style START fill:#e8f5e9,stroke:#2e7d32

    style MERGE fill:#fff3e0,stroke:#e65100

    style MEM0 fill:#e3f2fd,stroke:#1565c0

触发条件

InsightSelector 对每个用户检查三条触发路径，满足任意一条即纳入当晚洞察批次：

1. 1. 穿搭增量：距上次穿搭洞察后，新增穿搭记录数 >= insight_outfit_delta_threshold（默认 20）
2. 2. 衣橱增量：距上次衣橱洞察后，新增 source in (wardrobe, wardrobe_want) 且 status != disposed 的单品数 >= insight_item_delta_threshold（默认 20）
3. 3. 聊天到期：距上次聊天洞察 >= insight_chat_interval_days（默认 30 天）

首次洞察的特殊处理：若用户从未执行过洞察，只要累计穿搭数或衣橱单品数达到阈值，或近 30 天有聊天记录，均可触发首轮基线分析。

Sources: app/insights/selectors.py, app/insights/config.py

三维分析器

InsightAnalyzers 封装了四类 LLM 分析器，每个分析器使用专属 Prompt 模板处理对应维度数据：

聊天分析器提取 9 个核心维度：

每个观察事件都附带 置信度等级（High/Medium/Low），由 LLM 根据用户用词强度自动评估，作为后续仲裁权重参考。

Sources: app/insights/analyzers.py, data/prompts/memory/insight_chat.zh.txt

冲突仲裁引擎

冲突仲裁是离线链路中最关键的一步。analyze_merge 将三维洞察输出与用户当前 Mem0 记忆一同送入仲裁 Prompt，由 LLM 执行 7 大冲突法则，输出标准化的 add/update/delete 动作列表：

Mem0Synchronizer 负责解析仲裁输出并执行实际的 Mem0 操作：

flowchart LR

    RAW[LLM 仲裁输出<br/>yaml 标签] --> PARSE[parse_actions<br/>YAML 解析校验]

    PARSE --> VALID{校验通过?}

    VALID --> |否| ERR[抛出 ValueError]

    VALID --> |是| LOAD[加载当前记忆]

    LOAD --> CHECK{memory_id<br/>归属校验}

    CHECK --> |通过| EXEC[执行动作]

    EXEC --> ADD[mem.add]

    EXEC --> UPD[mem.update]

    EXEC --> DEL[mem.delete]

    CHECK --> |不归属| ERR2[抛出 ValueError]

Sources: data/prompts/memory/insight_merge.zh.txt, app/insights/mem0_sync.py

记忆分类与管理

自动分类系统

app/memory/admin.py 实现了基于关键词匹配的记忆分类机制，将每条记忆归入 7 个类别之一，用于 Dashboard 展示和管理筛选：

分类逻辑通过 classify_memory() 函数实现，对记忆文本进行小写归一化后逐一匹配关键词列表。other 类别的 SQL 查询策略是排除所有已知关键词，即"不属于任何已知类别"。

Sources: app/memory/admin.py

Admin API

管理员可通过 REST API 管理 Mem0 记忆，所有接口位于 /api/admin/memories 路径下，需要 Bearer Token 认证：