OpenSpace 实战：让 AI Agent 用 Skills、MCP 与谱系持续复用经验

一句话结论： OpenSpace 不是再造一个聊天 Agent，而是在现有 Agent 之上增加一层“技能管理系统”：任务执行后沉淀技能，用 SQLite 记录版本与谱系，在相似任务中复用，再通过 MCP 接入 Claude Code、Codex 等宿主。

>

适合读者： 已会使用 Python 和大模型 API，希望验证“任务执行 → 技能沉淀 → 相似任务复用 → 外部 Agent 调用”闭环的开发者。

>

来源： Sana Hassan，MarkTechPost，2026-07-25，*Building Self-Evolving AI Agents with OpenSpace Using Skills, MCP, Lineage, and Low-Cost Reuse*

>

项目： HKUDS/OpenSpace（MIT；官方 README 要求 Python 3.12+）

1. 先理解目标：这里的“自我演进”到底是什么

普通 Agent 每次都从提示词重新开始；OpenSpace 把执行中形成的可复用做法保存成 Skill，并记录它从哪里来、如何变化、是否被再次使用。

本文把闭环拆成四层：

  1. 执行层：通过 OpenSpace Python API 完成任务。
  2. 技能层：任务后分析结果，生成或更新 SKILL.md。
  3. 谱系层：在 SQLite 中记录技能版本、来源和演进关系。
  4. 接入层：通过 MCP 把技能管理能力暴露给 Claude Code、Codex 等宿主 Agent。

原文把技能来源概括为：

要注意：“存下了技能”不等于“系统真的变强了”。至少还要观察相似任务是否命中旧技能、结果质量是否稳定、Token 或执行成本是否下降，以及错误经验能否被淘汰。

2. 环境与安全边界

本教程沿用原文的 Google Colab 路线，便于快速验证，不直接改生产 Agent。

2.1 前置条件

2.2 不要把 Key 写进 Notebook

原文示例把 Key 留作空字符串。实际使用时，建议通过 Colab Secrets 或环境变量注入。

实践扩展：安全读取环境变量

# Colab：先在左侧 Secrets 中配置 ANTHROPIC_API_KEY 或 OPENAI_API_KEY
import os

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY", "")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
OPENSPACE_CLOUD_KEY = os.getenv("OPENSPACE_API_KEY", "")
OPENSPACE_MODEL = os.getenv(
    "OPENSPACE_MODEL",
    "anthropic/claude-sonnet-4-5",
)

if not (ANTHROPIC_API_KEY or OPENAI_API_KEY):
    print("未检测到 LLM Key：可以完成安装和数据库检查，但跳过真实任务。")

验收标准： 输出中不得出现完整 Key；分享 Notebook 前检查所有单元格输出。

3. 第一步：安装 OpenSpace 并建立隔离工作区

下面的单元格完成 Python 版本检查、稀疏克隆、可编辑安装和目录初始化。

import os
import pathlib
import shutil
import subprocess
import sys

assert sys.version_info >= (3, 12), (
    f"OpenSpace 需要 Python 3.12+，当前为 {sys.version.split()[0]}"
)

REPO_DIR = "/content/OpenSpace"
WORKSPACE = "/content/openspace_workspace"
SKILLS_DIR = "/content/my_agent_skills"

if not os.path.exists(REPO_DIR):
    subprocess.run(
        [
            "git", "clone", "--filter=blob:none", "--sparse",
            "https://github.com/HKUDS/OpenSpace.git", REPO_DIR,
        ],
        check=True,
    )
    subprocess.run(
        ["git", "sparse-checkout", "set", "--no-cone", "/*", "!/assets/"],
        cwd=REPO_DIR,
        check=True,
    )

subprocess.run(
    [sys.executable, "-m", "pip", "install", "-q", "-e", REPO_DIR],
    check=True,
)
subprocess.run(
    [sys.executable, "-m", "pip", "install", "-q", "nest_asyncio"],
    check=True,
)

os.makedirs(WORKSPACE, exist_ok=True)
os.makedirs(SKILLS_DIR, exist_ok=True)

for cli in ["openspace-mcp", "openspace-dashboard"]:
    print(cli, "=>", shutil.which(cli) or "未找到")

将非敏感配置写入 .env；Key 仍从环境变量读取：

config = {
    "OPENSPACE_MODEL": OPENSPACE_MODEL,
    "OPENSPACE_WORKSPACE": WORKSPACE,
    "OPENSPACE_HOST_SKILL_DIRS": SKILLS_DIR,
}

for key, value in config.items():
    os.environ[key] = value

safe_env = "\n".join(f"{k}={v}" for k, v in config.items()) + "\n"
pathlib.Path(REPO_DIR, "openspace", ".env").write_text(safe_env)
print("配置完成：", config)

预期输出形态：

openspace-mcp => /usr/local/bin/openspace-mcp
openspace-dashboard => /usr/local/bin/openspace-dashboard
配置完成： {'OPENSPACE_MODEL': '...', 'OPENSPACE_WORKSPACE': '...', ...}

失败处理：

4. 实战一：执行首个任务并检查技能是否产生

4.1 调用异步 Python API

import asyncio
import nest_asyncio

nest_asyncio.apply()

async def run_task(query: str):
    from openspace import OpenSpace

    async with OpenSpace() as agent:
        result = await agent.execute(query)

    print("── RESPONSE ──")
    print(result["response"][:3000])

    evolved = result.get("evolved_skills", [])
    for skill in evolved:
        print(f"Evolved: {skill['name']} (origin={skill['origin']})")
    return result

result_1 = asyncio.run(run_task(
    "Write a Python function that parses a CSV of employee hours and "
    "computes weekly payroll with overtime (1.5x beyond 40h). "
    "Test it on a small synthetic example and show the output."
))

4.2 你应该检查什么

不要只看自然语言答案；至少检查：

预期输出形态（名称会随任务变化）：

── RESPONSE ──
...工资计算代码与测试输出...
Evolved: <skill-name> (origin=CAPTURED)

如果 evolved_skills 为空，不要强行断言演进失败：可能是任务没有达到捕获阈值、后处理未完成，或当前版本配置不同。先查看日志和数据库。

5. 实战二：用相似任务验证“热复用”

第二个任务故意继承第一个任务的工资计算上下文，但新增税率 CSV 和净工资要求：

result_2 = asyncio.run(run_task(
    "Extend the payroll logic: add a second CSV of tax withholding rates "
    "per employee and produce net pay. Reuse any prior payroll skill."
))

5.1 查看 SQLite 技能库

import os
import sqlite3

runtime_db = os.path.join(WORKSPACE, ".openspace", "openspace.db")
repo_db = os.path.join(REPO_DIR, ".openspace", "openspace.db")
db_path = runtime_db if os.path.exists(runtime_db) else repo_db

if not os.path.exists(db_path):
    raise FileNotFoundError(f"没有找到 OpenSpace 数据库：{db_path}")

with sqlite3.connect(db_path) as conn:
    tables = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    ).fetchall()
    print("tables:", [row[0] for row in tables])

    for (table,) in tables:
        columns = conn.execute(f"PRAGMA table_info({table})").fetchall()
        print(table, "=>", [column[1] for column in columns])

这里只执行元数据查询。不要把来自用户输入的表名直接拼接进 SQL；上例的表名来自数据库自身的 sqlite_master。

5.2 热复用的验收标准

一次演示不能证明“成本持续下降”。建议把结果记录成对照表：

指标 首次任务 相似任务 判定
是否检索到已有技能 否/未知 是/否 相似任务应可命中
新技能来源 CAPTURED/FIX/其他 DERIVED/其他 是否形成谱系
输出是否通过测试 是/否 是/否 质量不能因复用下降
Token/费用 实测值 实测值 多轮后再判断趋势
总耗时 实测值 实测值 排除网络抖动后比较

本文提炼： “低成本复用”应视为待测假设，而不是看到数据库里有 Skill 就自动成立。最小可信证据是：同类任务命中旧技能、结果通过同一测试，并在多次重复中出现稳定的 Token 或耗时改善。

6. 实战三：手写一个 SKILL.md 并让 Agent 自动发现

6.1 创建 CSV 报告技能

import pathlib
import textwrap

skill_dir = pathlib.Path(SKILLS_DIR) / "colab-csv-report"
skill_dir.mkdir(parents=True, exist_ok=True)

(skill_dir / "SKILL.md").write_text(textwrap.dedent("""\
---
name: colab-csv-report
description: Turn any CSV into a short markdown report with summary stats, null counts, dtypes, and 3 key observations. Use pandas; never plot.
---
# colab-csv-report
1. Load the CSV with pandas (`on_bad_lines="skip"` fallback).
2. Emit: shape, dtypes table, describe(), null counts.
3. Write 3 bullet observations in plain Markdown.
4. If parsing fails, retry with `sep=None, engine="python"`.
"""))

print(skill_dir / "SKILL.md")

6.2 准备样本输入并调用技能

sample_csv = pathlib.Path("/content/demo.csv")
sample_csv.write_text(
    "name,dept,hours,rate\n"
    "Ada,Eng,45,90\n"
    "Grace,Eng,38,95\n"
    "Alan,Math,50,80\n"
)

csv_result = asyncio.run(run_task(
    f"Using the colab-csv-report skill, produce a markdown report for {sample_csv}"
))

预期输出至少包含： 数据形状、字段类型、描述统计、空值计数、3 条观察。

6.3 反向边界案例：空文件

empty_csv = pathlib.Path("/content/empty.csv")
empty_csv.write_text("")

empty_result = asyncio.run(run_task(
    f"Using the colab-csv-report skill, inspect {empty_csv}. "
    "Do not invent statistics; report that the input is empty."
))

验收标准： Agent 明确报告空输入或解析失败，不生成虚假的行数、均值或观察结论。

7. 通过 MCP 接入外部 Agent

原文使用 streamable HTTP 启动 MCP 服务：

import subprocess
import time
import urllib.error
import urllib.request

mcp_proc = subprocess.Popen(
    [
        "openspace-mcp",
        "--transport", "streamable-http",
        "--host", "127.0.0.1",
        "--port", "8081",
    ],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    env={**os.environ},
)

time.sleep(8)

try:
    request = urllib.request.Request("http://127.0.0.1:8081/mcp", method="GET")
    urllib.request.urlopen(request, timeout=5)
    print("MCP endpoint is reachable")
except urllib.error.HTTPError as error:
    print(f"MCP server is alive; bare GET returned HTTP {error.code}")
finally:
    mcp_proc.terminate()

裸 GET 返回非 2xx 不一定表示服务失效，因为 MCP 端点可能要求协议化请求。更可靠的验收是：进程存活、端口可达，并由真实 MCP 客户端完成一次工具发现。

宿主配置示意：

{
  "mcpServers": {
    "openspace": {
      "command": "openspace-mcp",
      "toolTimeout": 600,
      "env": {
        "OPENSPACE_HOST_SKILL_DIRS": "/content/my_agent_skills",
        "OPENSPACE_WORKSPACE": "/content/openspace_workspace"
      }
    }
  }
}

安全边界：

8. 如何阅读谱系，而不是只看技能数量

原文还检查了仓库 showcase/.openspace/openspace.db，并按 origin 统计技能。你可以使用同样的只读方法：

showcase_db = os.path.join(REPO_DIR, "showcase", ".openspace", "openspace.db")

if os.path.exists(showcase_db):
    with sqlite3.connect(showcase_db) as conn:
        for (table,) in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        ):
            columns = [
                row[1].lower()
                for row in conn.execute(f"PRAGMA table_info({table})")
            ]
            if "origin" in columns:
                rows = conn.execute(
                    f"SELECT origin, COUNT(*) FROM {table} "
                    "GROUP BY origin ORDER BY COUNT(*) DESC"
                ).fetchall()
                print(table, rows)

分析时重点问四个问题：

  1. 来源是否清楚？ 能否区分 CAPTURED、DERIVED、FIX？
  2. 父子关系是否可追踪？ 派生技能能否找到父技能和版本？
  3. 质量是否有证据？ 是否只有“生成记录”，没有任务结果或测试？
  4. 错误能否退出？ 低质量技能是否可以停用、回滚或被更高质量版本替代？

技能数量增长不是目标；可审计的质量增长才是目标。

9. 一套更可靠的落地顺序

这是基于原文流程整理出的实践扩展，不是原文声称的官方实施规范：

阶段 A：只读观察

阶段 B：受控复用

阶段 C：宿主接入

阶段 D：团队共享

10. 常见失败与排查

没有生成技能

检查任务是否完成、后处理是否运行、日志是否报错，以及当前版本是否改变了演进阈值或字段。不要通过反复改提示词制造“看起来像演进”的输出。

相似任务没有复用旧技能

检查 OPENSPACE_HOST_SKILL_DIRS 与 OPENSPACE_WORKSPACE 是否在同一进程中生效；确认技能目录可读；再检查检索日志和技能描述是否足以匹配任务。

MCP 端口可达但客户端找不到工具

裸 HTTP 探针只能证明服务进程响应。继续检查客户端配置、传输类型、环境变量、工作目录和 MCP 初始化日志。

Token 没有下降

单次对比没有统计意义。固定模型、输入、工具和测试，至少进行多轮冷/热对照；同时确认复用没有以牺牲质量为代价。

技能越积越多，效果反而变差

这是技能库的典型维护风险。需要去重、质量门槛、停用机制和谱系回滚，而不是继续增加技能数量。

11. 原文方法覆盖清单

原文方法 是否纳入 落点 若遗漏原因
Python 3.12+、稀疏克隆、可编辑安装 是 第 2～3 节 —
配置模型、工作区与技能目录 是 第 2～3 节 Key 改为环境变量注入
异步 Python API 执行任务 是 第 4 节 —
检查 evolved skills 与 SQLite 是 第 4～5 节 —
相似工资任务验证复用 是 第 5 节 —
创建自定义 SKILL.md 是 第 6 节 —
安装 delegate-task、skill-discovery 宿主技能 否 — 它依赖当前仓库目录结构，建议按官方 host integration 文档执行，避免复制过时路径
启动 streamable HTTP MCP 是 第 7 节 增加 loopback 与协议探针边界
云端技能上传/下载 部分 第 7、9 节 未执行外部上传；只保留安全检查与阶段建议
检查 showcase DB 和 FIX/DERIVED/CAPTURED 是 第 8 节 —
Dashboard 否 — 不是验证技能演进闭环的必要步骤，且还需 Node 20+ 与前端安装

12. 最终验收清单

完成教程后，你应该能回答“是”或给出明确失败原因：

结语

OpenSpace 最值得验证的不是“Agent 会自动写 Skill”，而是它能否把任务经验变成可检索、可测试、可追踪、可回滚的资产。先在低风险任务上建立冷/热对照，再接 MCP、团队共享和自动演进；否则，系统可能只是把临时提示词变成了一个不断膨胀的技能仓库。

证据与验证状态