企业官网建设流程全解析

OpenViking Agent Evolution API 实战指南：Experience 应用轨迹查询与结果分布统计

【免费下载链接】OpenVikingSelf-evolving Context Database for AI Agents. Unify Agent Memory, Knowledge RAG and Skills.项目地址: https://gitcode.com/GitHub_Trending/op/OpenViking

本篇技术指南围绕 OpenViking 的Agent Evolution（智能体演进）API展开，聚焦两个 HTTP 接口：按 Experience 列出应用轨迹（trajectories）与统计 Experience 的五类结果分布（outcome distribution）。通过本文，读者将掌握这两个接口的完整调用方式、参数语义、响应结构，以及其底层基于标量标签（scalar tag）聚合的实现原理，并能在自己的 Agent 记忆分析、策略调优场景中直接复用这套"消费了哪条经验、效果如何"的可观测闭环。

Agent Evolution 是什么

OpenViking 是一个面向 AI Agent 的自演进上下文数据库（Self-evolving Context Database），统一管理 Agent 的 Memory、Knowledge RAG 与 Skills。Agent Evolution 是其会话记忆体系中的一环：当一次会话提交（commit）时，若会话读取（read）了某条 Experience（经验），系统会生成对应的trajectory（轨迹）记录，并给轨迹打上结果标签（如success、failure）。

Agent Evolution API 正是用来回答两个核心问题的：

• 某条 Experience 被哪些会话应用过？—— 列出消费了该 Experience 的轨迹；
• 应用的最终效果如何？—— 统计这些轨迹的结果分布。

这两个接口目前仅通过 HTTP 提供（文档原文明确说明 "currently available through HTTP only"），查询范围严格限定为当前用户自己拥有的 Experience 与轨迹，具备天然的多租户隔离语义。完整的 HTTP 路由定义见 agent_evolution.py，其路由前缀为/api/v1/agent-evolution。

背景：轨迹与结果标签从何而来

在深入接口之前，先理解数据是如何产生的，这有助于正确解读查询结果。

会话提交时生成轨迹

会话提交（commit）流程会从消息集中提取"成功读取过的 Experience URI"，并据此生成轨迹记忆。核心逻辑位于 experience_lineage.py 的collect_read_experience_uris()函数：

• 它会扫描会话消息中的ToolPart，识别read、multi_read、openviking_read、ov_read以及mcp__openviking__read等读取类工具调用；
• 仅统计工具状态为completed（成功完成）的调用；
• 通过canonical_experience_uri()校验 URI 属于当前用户且是 Experience 文件；
• 通过_read_failed_in_text()进一步排除输出中标记为失败（如(nothing found at {uri})或ERROR:段落）的读取。

也就是说，只有真正成功消费过某条 Experience 的会话才会产生对应的轨迹，这保证了后续统计的语义纯净。

轨迹的存储位置

轨迹文件存放在用户命名空间下的固定目录：

viking://user/{user_id}/memories/trajectories

这一点在服务层 agent_evolution_service.py 中有明确体现：trajectory_root = f"viking://user/{ctx.user.user_id}/memories/trajectories"。查询时通过PathScope("uri", trajectory_root, depth=1)将检索范围锁定在该目录下。

五类结果标签

轨迹的结果（outcome）通过标量标签trajectory_outcome={outcome}标记，取值由 experience_lineage.py 中的常量定义：

TRAJECTORY_OUTCOMES = ("success", "failure", "partial", "unknown", "unfinished")

训练侧组件 domain.py 以TrajectoryOutcome = Literal["success", "failure", "partial", "unfinished", "unknown"]的形式复用了同一组取值。需要特别注意的是文档中的关键限制：

Trajectories created by older versions and not yet re-indexed do not carry outcome tags and are therefore excluded（旧版本创建的、尚未重新索引的轨迹不带结果标签，因此会被排除在统计之外）。

接口一：列出 Experience 应用轨迹

接口定义

GET /api/v1/agent-evolution/experiences/trajectories

该接口返回分页的、成功读取过指定 Experience 的轨迹列表。HTTP 路由入口为openviking/server/routers/agent_evolution.py:list_experience_trajectories，核心实现为openviking/service/agent_evolution_service.py:AgentEvolutionService.list_trajectories_by_experience。

参数说明

参数约束在路由层与服务层都有严格校验。路由层 agent_evolution.py 通过 FastAPI 的Query(ge=1, le=MAX_TRAJECTORY_PAGE_LIMIT)限定limit范围（MAX_TRAJECTORY_PAGE_LIMIT = 1000）、offset >= 0；服务层 agent_evolution_service.py 会再次校验并抛出InvalidArgumentError。API 测试 test_api_agent_evolution.py 验证了limit=1001时返回 400。

调用示例

curl -X GET "http://localhost:1933/api/v1/agent-evolution/experiences/trajectories?experience_uri=viking://user/default/memories/experiences/exchange.md&limit=50&offset=0&start_date=2026-08-01&end_date=2026-08-10" \ -H "X-API-Key: your-key"

响应示例

{ "status": "ok", "result": { "experience_uri": "viking://user/default/memories/experiences/exchange.md", "items": [ { "uri": "viking://user/default/memories/trajectories/exchange_20260805020000.md", "name": "exchange_20260805020000.md", "description": "Handle an exchange request", "created_at": "2026-08-05T02:00:00Z", "updated_at": "2026-08-05T02:00:00Z" } ], "total": 1, "limit": 50, "offset": 0, "has_more": false }, "time": 0.01 }

响应中的每个 item 仅包含索引字段中实际存在的字段，即uri、name、description、created_at、updated_at五个字段的子集。这在服务层通过 agent_evolution_service.py 的字段投影逻辑实现：{field: record.get(field) for field in _TRAJECTORY_OUTPUT_FIELDS if field in record}。

has_more字段表示是否还有更多数据，其计算方式为offset + len(items) < total。服务测试 test_agent_evolution_service.py 验证了limit=1000时分页拉取 1001 条记录的场景，has_more从true正确翻转为false。

底层实现：精确标量过滤与分页

list_trajectories_by_experience的实现要点如下：

1. 校验与规范化：通过canonical_experience_uri()将传入 URI 规范化为viking://user/{user_id}/memories/experiences/...形态，并强制校验属于当前用户；随后用viking_fs.stat()确认它指向文件而非目录（agent_evolution_service.py）；
2. 构造过滤条件（_experience_trajectory_conditions）：
   • PathScope("uri", trajectory_root, depth=1)—— 限定轨迹目录；
   • Eq("context_type", "memory")与Eq("level", 2)—— 限定 memory 类、L2 级别；
   • Eq("search_tags", experience_source_tag(experience_uri))—— 精确匹配"该 Experience 被读取"的源标签；
3. 日期区间过滤：_trajectory_created_at_range()将YYYY-MM-DD解析为 UTC 时间区间。这里有一个实现细节值得注意：底层TimeRange使用开区间上界，因此实现通过end + timedelta(days=1)将end_date转换为次日零点，从而保证end_date当天包含在内（agent_evolution_service.py）。对应测试 test_agent_evolution_service.py 验证了start_date=2026-08-01, end_date=2026-08-10被转换为start="2026-08-01T00:00:00+00:00", end="2026-08-11T00:00:00+00:00"；
4. 并发查询：asyncio.gather同时发起vikingdb.filter()（取记录）与vikingdb.count()（取总数），按updated_at降序排序。

experience_source_tag()是这条链路的关键：它把 Experience URI 编码成严格的k=v标签（形如...=1），并通过_escape_search_tag_key()对 URI 中的大写字符、%、=做百分号转义，保证 URI 身份在标签归一化（小写化）过程中不被破坏（experience_lineage.py）。

接口二：获取 Experience 结果分布

接口定义

GET /api/v1/agent-evolution/experiences/outcomes

该接口统计消费过指定 Experience 的轨迹在五种结果上的数量分布。文档强调：查询使用精确标量标签聚合，不会加载每个轨迹文件，因此即使轨迹数量很大，统计成本也很低。HTTP 路由入口为openviking/server/routers/agent_evolution.py:get_experience_outcome_distribution，核心实现为AgentEvolutionService.get_experience_outcome_distribution。

参数说明

调用示例

curl -X GET "http://localhost:1933/api/v1/agent-evolution/experiences/outcomes?experience_uri=viking://user/default/memories/experiences/exchange.md&start_date=2026-08-01&end_date=2026-08-10" \ -H "X-API-Key: your-key"

响应示例

{ "status": "ok", "result": { "experience_uri": "viking://user/default/memories/experiences/exchange.md", "outcome_distribution": [ {"outcome": "success", "count": 4}, {"outcome": "failure", "count": 1}, {"outcome": "partial", "count": 0}, {"outcome": "unknown", "count": 0}, {"outcome": "unfinished", "count": 0} ] }, "time": 0.01 }

响应中的outcome_distribution始终包含全部五个结果：success、failure、partial、unknown、unfinished（数量为 0 也会列出）。这是由服务层用zip(TRAJECTORY_OUTCOMES, counts, strict=True)严格配对实现的（agent_evolution_service.py），对下游做图表渲染或阈值告警非常友好。

底层实现：五路并发精确计数

get_experience_outcome_distribution的实现思路是"一次过滤、五路计数"：

1. 与轨迹列表接口相同，先完成 URI 规范化、用户归属校验、文件类型校验（_prepare_experience_query）；
2. 构造基础过滤条件（同上：目录范围 + memory 类型 + L2 级别 + Experience 源标签，可选日期区间）；
3. 对五种 outcome 分别追加Eq("search_tags", trajectory_outcome_tag(outcome))条件，并通过asyncio.gather并发执行 5 次vikingdb.count()；
4. 按固定顺序组装结果。

其中trajectory_outcome_tag()生成形如trajectory_outcome=success的严格标量标签，且经过normalize_trajectory_outcome()归一化——任何非五类取值都会回退为unknown（experience_lineage.py），保证了标签集合的封闭性。服务测试 test_agent_evolution_service.py 验证了同一条轨迹上search_tags同时携带源标签与结果标签时，两种条件能正确组合计数。

结果的语义边界

使用结果分布时务必注意文档明确的两点：

• 五种结果固定返回，不会因为某类数量为 0 而省略；
• 旧版本创建且尚未重新索引的轨迹不带trajectory_outcome标签，会被排除，因此统计结果反映的是"当前已索引数据"的分布，若需完整口径应确保数据已重索引。

实战场景：用结果分布驱动经验迭代

把两个接口组合起来，可以构建一个经典的"经验效果评估闭环"：

1. 发现候选：通过轨迹列表接口，查看某条 Experience（如viking://user/default/memories/experiences/exchange.md）近期被哪些会话应用、应用频率如何；
2. 评估效果：通过结果分布接口，统计该 Experience 的success与failure比例。例如"成功率 4/5、unfinished为 0"说明经验质量良好；若failure或unfinished占比升高，则提示该经验需要更新或淘汰；
3. 时间切片：结合start_date/end_date对比不同时间窗口（如版本发布前后）的成功率变化，验证经验迭代是否生效。

工程实现要点与验证

服务注册

Agent Evolution 路由通过 app.py 的app.include_router(agent_evolution_router)挂载进 FastAPI 应用，路由前缀/api/v1/agent-evolution与tags=["agent-evolution"]定义在 agent_evolution.py。鉴权走统一请求上下文get_request_context()，API Key 通过X-API-Key请求头传递。

部署级开关

Agent Evolution 的生成受部署级全局开关控制（默认关闭），配置于服务端server.agent_evolution.enabled：

{ "server": { "agent_evolution": { "enabled": false } } }

开关关闭时：会话提交不会新生成或更新cases、trajectories、experiences三类记忆，但不会删除已有文件，也不会阻止已有 Experience 被搜索和读取。会话级memory_policy在开关开启时仍是权威的"允许列表"；开关关闭时则无法通过memory_policy绕过。完整设计见 agent-evolution-global-switch-design.md。也就是说，若部署未开启该开关，即使调用本文的两个接口，也可能查询不到新生成的轨迹数据——这是排查"接口返回为空"时的第一检查项。

测试覆盖

仓库为这两个接口提供了双层测试：

• 服务层单测test_agent_evolution_service.py：覆盖精确过滤条件构造、分页边界（limit=1000、offset 翻页）、日期区间闭区间语义、非法日期区间（start_date > end_date）、越界 limit（>1000）、跨用户 URI 拒绝等；
• HTTP 层测试test_api_agent_evolution.py：验证默认分页（limit=50、offset=0）、日期参数透传、limit=1001返回 400、结果分布接口的响应结构。

小结

OpenViking 的 Agent Evolution API 用两个只读 HTTP 接口，为"经验 → 应用 → 效果"提供了精确、低成本、多租户隔离的可观测能力：

其底层依赖"提交时记录读取来源 + 打结果标签"的写入侧机制（experience_lineage.py）与"严格标签过滤 + 并发计数"的查询侧实现（agent_evolution_service.py），两者共同构成了 Agent 经验从沉淀到评估的完整闭环。

相关文档

• Sessions（提交会话并生成 Agent Evolution 记忆）
• Memory（读取与召回记忆）
• Agent Evolution 全局开关设计

【免费下载链接】OpenVikingSelf-evolving Context Database for AI Agents. Unify Agent Memory, Knowledge RAG and Skills.项目地址: https://gitcode.com/GitHub_Trending/op/OpenViking