# 在 Colab 搭建一个可测试的 QwenPaw Agent 工作区

> 把 MarkTechPost 的 QwenPaw 示例提炼成一篇可执行教程：从安装、目录规划、模型供应商配置、自定义 skill、本地知识文件，到 Console 启动和流式 API 测试。

## 来源与边界

- 原文标题：How to Build a QwenPaw Agent Workspace with Custom Skills, Model Providers, Console Access, and Streaming API Testing
- 原文链接：https://www.marktechpost.com/2026/06/13/how-to-build-a-qwenpaw-agent-workspace-with-custom-skills-model-providers-console-access-and-streaming-api-testing/
- 作者：Sana Hassan
- 发布时间：2026-06-13
- 官方参考：QwenPaw GitHub / PyPI 当前公开信息显示快速启动路径为 `pip install qwenpaw`、`qwenpaw init --defaults`、`qwenpaw app`，Console 默认地址为 `http://127.0.0.1:8088/`。
- 原文未提供可验证 GitHub 教程仓库或完整 Notebook URL；下文代码是基于原文步骤整理的教学版，不直接复制原文长代码。

## 你会搭建什么

完成后，你会得到一个 Colab 里的 QwenPaw 实验工作区：

```text
/content/qwenpaw_colab/
├── working/                         # QwenPaw 工作目录
│   └── workspaces/default/           # 默认 Agent 工作区
│       ├── skills/research_brief/     # 自定义技能
│       │   └── SKILL.md
│       ├── demo_knowledge/
│       │   └── qwenpaw_colab_notes.md
│       └── README_COLAB_TUTORIAL.md
├── secrets/                          # 凭证目录，不提交、不公开
├── logs/                             # 服务日志
├── qwenpaw_app.pid                   # 进程 PID
└── logs/qwenpaw_app.log              # app 日志
```

能力目标：

1. 在 Colab 安装并初始化 QwenPaw。
2. 用环境变量或 Colab secrets 配置模型供应商。
3. 创建一个默认 Agent：`Colab Research Assistant`。
4. 添加一个 `research_brief` 自定义 skill。
5. 放入本地知识文件，让 Agent 能读取 workspace 上下文。
6. 启动 Console。
7. 通过 REST API 做一次流式调用验证。

## 适用场景

适合：

- 快速试用 QwenPaw Agent 工作区。
- 给团队演示“Agent + 本地文件 + 自定义技能 + API 调用”的最小闭环。
- 做一次性 Colab PoC。

不适合直接用于生产：

- Colab runtime 会重置，不适合长期服务。
- 公网 tunnel 有访问暴露风险。
- API key、访问凭据、日志和 workspace 内容必须严格隔离。
- 高并发、审计、权限分级、持久化部署需要另行设计。

---

## 第 0 步：准备模型凭证

任选一个模型供应商即可。推荐在 Colab secrets 或环境变量中设置，不要写死在 Notebook 代码里。

支持的变量名示例：

```text
OPENAI_API_KEY
OPENROUTER_API_KEY
DASHSCOPE_API_KEY
DEEPSEEK_API_KEY
GEMINI_API_KEY
GOOGLE_API_KEY
```

如果没有模型凭证，Console 仍可能启动，但无法正常聊天。

## 第 1 步：安装依赖并建立目录

在 Colab 新建一个 Notebook，先运行下面这个 cell。

```python
from __future__ import annotations

import json
import os
import pathlib
import secrets
import shutil
import socket
import subprocess
import sys
import time
from dataclasses import dataclass

PORT = int(os.environ.get("QWENPAW_COLAB_PORT", "8088"))
ROOT = pathlib.Path("/content/qwenpaw_colab")
WORKING_DIR = ROOT / "working"
SECRET_DIR = ROOT / "secrets"
LOG_DIR = ROOT / "logs"
WORKSPACE_DIR = WORKING_DIR / "workspaces" / "default"
PID_FILE = ROOT / "qwenpaw_app.pid"
APP_LOG = LOG_DIR / "qwenpaw_app.log"

for path in [ROOT, WORKING_DIR, SECRET_DIR, LOG_DIR, WORKSPACE_DIR]:
    path.mkdir(parents=True, exist_ok=True)

os.environ["QWENPAW_WORKING_DIR"] = str(WORKING_DIR)
os.environ["QWENPAW_SECRET_DIR"] = str(SECRET_DIR)
os.environ["QWENPAW_AUTH_ENABLED"] = "true"
os.environ["QWENPAW_AUTH_USERNAME"] = os.environ.get("QWENPAW_AUTH_USERNAME", "admin")
os.environ["QWENPAW_LOG_LEVEL"] = os.environ.get("QWENPAW_LOG_LEVEL", "info")
os.environ["QWENPAW_SKILL_SCAN_MODE"] = os.environ.get("QWENPAW_SKILL_SCAN_MODE", "warn")
os.environ["QWENPAW_TOOL_GUARD_ENABLED"] = os.environ.get("QWENPAW_TOOL_GUARD_ENABLED", "true")

password_file = SECRET_DIR / ".colab_ui_password"
if not password_file.exists():
    password_file.write_text("qpw-" + secrets.token_urlsafe(18), encoding="utf-8")
os.environ["QWENPAW_AUTH_PASSWORD"] = password_file.read_text(encoding="utf-8").strip()

print("Python:", sys.version)
print("Workspace:", WORKSPACE_DIR)
print("Console user:", os.environ["QWENPAW_AUTH_USERNAME"])
print("Console password saved at:", password_file)
```

预期输出形态：

```text
Python: 3.x.x (...)
Workspace: /content/qwenpaw_colab/working/workspaces/default
Console user: admin
Console password saved at: /content/qwenpaw_colab/secrets/.colab_ui_password
```

安全提示：不要把密码输出截图发到公开群，也不要把 `/content/qwenpaw_colab/secrets` 打包分享。

## 第 2 步：安装 QwenPaw 并初始化

```python
def run_command(args: list[str], check: bool = False) -> subprocess.CompletedProcess[str]:
    """Run a command without shell expansion and print a compact tail."""
    print("\n$", " ".join(args))
    result = subprocess.run(
        args,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        check=False,
    )
    print(result.stdout[-4000:])
    if check and result.returncode != 0:
        raise RuntimeError(f"command failed: {args} rc={result.returncode}")
    return result


def qwenpaw_cmd(*args: str) -> list[str]:
    """Return qwenpaw executable command."""
    executable = shutil.which("qwenpaw")
    if executable:
        return [executable, *args]
    return [sys.executable, "-m", "qwenpaw", *args]


assert sys.version_info >= (3, 10), "QwenPaw needs Python 3.10+."
run_command([sys.executable, "-m", "pip", "install", "-q", "-U", "pip", "setuptools", "wheel"])
run_command([sys.executable, "-m", "pip", "install", "-q", "-U", "qwenpaw", "requests"], check=True)

config_path = WORKING_DIR / "config.json"
if not config_path.exists():
    run_command(qwenpaw_cmd("init", "--defaults"), check=False)
else:
    print("QwenPaw already initialized:", WORKING_DIR)
```

验证：

```python
run_command(qwenpaw_cmd("daemon", "version"), check=False)
```

如果 `qwenpaw` 命令找不到，先确认上一段安装是否成功；如果安装超时，重启 Colab runtime 后重跑前两步。

## 第 3 步：配置模型供应商

这一步做三件事：

1. 从环境变量或 Colab secrets 找可用 key。
2. 生成 provider 配置文件。
3. 写入 active model。

```python
@dataclass(frozen=True)
class ProviderCandidate:
    env_name: str
    provider_id: str
    name: str
    base_url: str
    model: str
    chat_model: str
    key_prefix: str = ""


def colab_secret_or_env(name: str) -> str:
    """Read a secret from environment or Google Colab userdata."""
    value = os.environ.get(name, "")
    try:
        from google.colab import userdata  # type: ignore[import-not-found]

        secret_value = userdata.get(name)
        if secret_value:
            value = secret_value
    except Exception:
        pass
    return value or ""


def read_json(path: pathlib.Path, default: dict) -> dict:
    """Read a JSON object or return default."""
    if not path.exists():
        return default
    return json.loads(path.read_text(encoding="utf-8"))


def write_json(path: pathlib.Path, data: dict) -> None:
    """Write JSON with parent directory creation."""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")


provider_candidates = [
    ProviderCandidate("OPENAI_API_KEY", "openai", "OpenAI", "https://api.openai.com/v1", "gpt-4o-mini", "OpenAIChatModel", "sk-"),
    ProviderCandidate("OPENROUTER_API_KEY", "openrouter", "OpenRouter", "https://openrouter.ai/api/v1", "openai/gpt-4o-mini", "OpenAIChatModel", "sk-or-"),
    ProviderCandidate("DASHSCOPE_API_KEY", "dashscope", "DashScope", "https://dashscope.aliyuncs.com/compatible-mode/v1", "qwen-plus", "OpenAIChatModel", "sk-"),
    ProviderCandidate("DEEPSEEK_API_KEY", "deepseek", "DeepSeek", "https://api.deepseek.com", "deepseek-chat", "OpenAIChatModel", "sk-"),
    ProviderCandidate("GEMINI_API_KEY", "gemini", "Google Gemini", "https://generativelanguage.googleapis.com", "gemini-2.5-flash", "GeminiChatModel"),
    ProviderCandidate("GOOGLE_API_KEY", "gemini", "Google Gemini", "https://generativelanguage.googleapis.com", "gemini-2.5-flash", "GeminiChatModel"),
]

selected: tuple[ProviderCandidate, str] | None = None
for candidate in provider_candidates:
    api_key = colab_secret_or_env(candidate.env_name)
    if api_key:
        selected = (candidate, api_key)
        break

if selected is None:
    print("No provider key found. Add one API key, then rerun this cell.")
else:
    candidate, api_key = selected
    provider_dir = SECRET_DIR / "providers" / "builtin"
    provider_payload = {
        "id": candidate.provider_id,
        "name": candidate.name,
        "base_url": candidate.base_url,
        "api_key": api_key,
        "chat_model": candidate.chat_model,
        "extra_models": [
            {
                "id": candidate.model,
                "name": candidate.model,
                "max_tokens": 2048,
                "max_input_length": 131072,
                "generate_kwargs": {"temperature": 0.2, "max_tokens": 2048},
            }
        ],
        "api_key_prefix": candidate.key_prefix,
        "require_api_key": True,
        "auth_mode": "api_key",
    }
    write_json(provider_dir / f"{candidate.provider_id}.json", provider_payload)
    write_json(
        SECRET_DIR / "providers" / "active_model.json",
        {"provider_id": candidate.provider_id, "model": candidate.model},
    )
    print("Configured provider:", candidate.name, candidate.model)
```

失败处理：

- 输出 `No provider key found`：说明没有配置任何 API key。
- 模型调用失败：优先检查 provider 的 `base_url`、模型名和 key 是否匹配。
- 不确定 provider 配置字段是否变化：以当前 QwenPaw 官方文档为准，本文字段来自原文教程与公开 quick start 信息。

## 第 4 步：创建默认 Agent 配置

```python
agent_dir = WORKING_DIR / "agents" / "default"
agent_path = agent_dir / "agent.json"
agent = read_json(agent_path, {})

agent.update(
    {
        "id": "default",
        "name": "Colab Research Assistant",
        "description": "QwenPaw agent for Colab: file-aware, skill-aware, API-testable, and guarded.",
        "language": "en",
        "workspace_dir": str(WORKSPACE_DIR),
        "enabled": True,
        "channels": {"console": {"enabled": True}},
        "running": {"max_iters": 30, "llm_retry_enabled": True, "stream_output": True},
        "security": {
            "tool_guard": True,
            "file_guard": True,
            "skill_scanner": True,
            "skill_scan_mode": "warn",
        },
        "memory": {"enabled": True},
    }
)

active_model_path = SECRET_DIR / "providers" / "active_model.json"
if active_model_path.exists():
    agent["active_model"] = read_json(active_model_path, {})

write_json(agent_path, agent)
print("Agent config written:", agent_path)
```

验证：

```python
print(agent_path.read_text(encoding="utf-8")[:1000])
```

至少应看到：

```json
{
  "id": "default",
  "name": "Colab Research Assistant",
  "workspace_dir": "/content/qwenpaw_colab/working/workspaces/default"
}
```

## 第 5 步：添加一个自定义 skill

这里把原文的 `research_brief` 思路改成一个更实用的团队研究简报技能。

```python
skill_dir = WORKSPACE_DIR / "skills" / "research_brief"
skill_dir.mkdir(parents=True, exist_ok=True)

skill_text = """---
name: research_brief
description: Create rigorous research briefs from user questions, local notes, uploaded files, and available tools.
---

# Research Brief Skill

Use this skill when the user asks for research, product analysis, market mapping,
technical due diligence, paper analysis, repo analysis, or a decision memo.

## Procedure

1. Restate the user's objective in one sentence.
2. Identify the most important entities, assumptions, and constraints.
3. Search available local workspace files first.
4. Use tools only when they are relevant and allowed.
5. Separate verified facts from inference.
6. Produce a compact brief with:
   - answer
   - evidence
   - risks or caveats
   - recommended next step

## Output Style

Prefer clear sections, short paragraphs, and explicit uncertainty.
Do not invent citations, file contents, commands, or results.
"""

(skill_dir / "SKILL.md").write_text(skill_text, encoding="utf-8")
print("Skill written:", skill_dir / "SKILL.md")
```

为什么这个 skill 有用：

- 它不直接让 Agent “自由发挥”。
- 它要求先看本地文件，再区分事实与推论。
- 它把输出固定为 answer / evidence / risks / next step，便于审查。

## 第 6 步：添加本地知识文件

```python
demo_dir = WORKSPACE_DIR / "demo_knowledge"
demo_dir.mkdir(parents=True, exist_ok=True)

(demo_dir / "qwenpaw_colab_notes.md").write_text(
    """# QwenPaw Colab Demo Notes

This workspace demonstrates:

- QwenPaw installation and initialization
- provider configuration from Colab secrets or environment variables
- authenticated Console launch
- custom workspace skill creation
- local workspace knowledge files
- streaming REST API calls
- optional public tunnel exposure

Recommended first prompt:
Read my workspace notes and explain what this QwenPaw Colab setup can do.
Then use the research_brief skill style to propose three advanced experiments.
""",
    encoding="utf-8",
)

(WORKSPACE_DIR / "README_COLAB_TUTORIAL.md").write_text(
    """# QwenPaw Advanced Colab Workspace

Suggested experiments:

1. Ask QwenPaw to inspect the demo_knowledge folder.
2. Ask it to use the research_brief skill style.
3. Use the REST API client for automated tests.
4. Add more SKILL.md folders under workspace/skills.
5. Add more notes, CSVs, markdown files, or task briefs under workspace folders.
""",
    encoding="utf-8",
)

print("Demo files:")
for path in sorted(WORKSPACE_DIR.rglob("*.md")):
    print("-", path)
```

预期输出：

```text
Demo files:
- .../README_COLAB_TUTORIAL.md
- .../demo_knowledge/qwenpaw_colab_notes.md
- .../skills/research_brief/SKILL.md
```

## 第 7 步：检查 QwenPaw 能否识别模型和技能

```python
run_command(qwenpaw_cmd("models", "list"), check=False)
run_command(qwenpaw_cmd("skills", "list", "--agent-id", "default"), check=False)
```

如果 `skills list` 没看到自定义 skill：

1. 检查 skill 文件路径是否是 `workspace/skills/research_brief/SKILL.md`。
2. 检查 YAML frontmatter 是否存在 `name` 和 `description`。
3. 检查 `QWENPAW_SKILL_SCAN_MODE` 是否过严；实验阶段可先用 `warn`。

## 第 8 步：启动 Console

```python
import signal


def port_open(host: str = "127.0.0.1", port: int = 8088, timeout: float = 0.5) -> bool:
    """Return True when the TCP port accepts connections."""
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def wait_for_port(port: int, seconds: int = 90) -> bool:
    """Wait until a port is open."""
    start = time.time()
    while time.time() - start < seconds:
        if port_open("127.0.0.1", port):
            return True
        time.sleep(1)
    return False


def stop_previous_app() -> None:
    """Stop a previously recorded QwenPaw app process."""
    if not PID_FILE.exists():
        return
    try:
        pid = int(PID_FILE.read_text(encoding="utf-8").strip())
        os.kill(pid, signal.SIGTERM)
        time.sleep(2)
    except Exception:
        pass
    PID_FILE.unlink(missing_ok=True)


stop_previous_app()
log_file = APP_LOG.open("w", encoding="utf-8")
app_proc = subprocess.Popen(
    qwenpaw_cmd("app", "--host", "0.0.0.0", "--port", str(PORT), "--log-level", os.environ["QWENPAW_LOG_LEVEL"]),
    stdout=log_file,
    stderr=subprocess.STDOUT,
    env=os.environ.copy(),
)
PID_FILE.write_text(str(app_proc.pid), encoding="utf-8")

if not wait_for_port(PORT, seconds=120):
    print(APP_LOG.read_text(encoding="utf-8", errors="ignore")[-6000:])
    raise RuntimeError("QwenPaw app failed to start")

print(f"QwenPaw app is running on http://127.0.0.1:{PORT}")
print("Username:", os.environ["QWENPAW_AUTH_USERNAME"])
print("Password file:", SECRET_DIR / ".colab_ui_password")
print("App log:", APP_LOG)
```

Colab proxy URL：

```python
try:
    from google.colab import output  # type: ignore[import-not-found]

    proxy_url = output.eval_js(f"google.colab.kernel.proxyPort({PORT})")
    print("Colab proxied Console URL:")
    print(proxy_url)
except Exception as exc:
    print("Not running inside Colab proxy environment:", exc)
```

安全建议：

- 默认只使用 Colab proxy。
- 临时公网 tunnel 只适合短时演示。
- 不要把公网 URL、用户名、密码同时发给不可信对象。
- 如果接入真实数据源，先关闭公网 tunnel。

## 第 9 步：用流式 API 做一次闭环测试

下面这个测试会调用 `/api/console/chat`。不同 QwenPaw 版本的 API 字段可能变化；如果失败，先看 Console 网络请求或官方 API 文档校正 endpoint 和 payload。

```python
import requests


def stream_chat(message: str) -> None:
    """Send a streaming chat request to the local QwenPaw console API."""
    url = f"http://127.0.0.1:{PORT}/api/console/chat"
    payload = {
        "agent_id": "default",
        "message": message,
        "stream": True,
    }
    with requests.post(url, json=payload, stream=True, timeout=120) as response:
        print("status:", response.status_code)
        response.raise_for_status()
        for line in response.iter_lines(decode_unicode=True):
            if line:
                print(line[:500])


stream_chat(
    "Read my workspace notes and explain what this QwenPaw Colab setup can do. "
    "Then use the research_brief skill style to propose three advanced experiments."
)
```

预期输出形态：

```text
status: 200
... workspace notes ...
... research brief ...
... risks or caveats ...
... recommended next step ...
```

如果失败：

- `Connection refused`：Console 没启动，回到第 8 步看日志。
- `401/403`：API 可能需要认证头或 cookie；先通过 Console 验证登录，再查当前版本 API 文档。
- `404`：接口路径可能变化，检查浏览器 DevTools 或官方文档。
- `5xx`：优先看 `APP_LOG`，再检查模型 provider 是否配置成功。

## 第 10 步：停止服务

```python
stop_previous_app()
print("Stopped QwenPaw app if it was running.")
```

## 最小验收清单

- [ ] `qwenpaw daemon version` 能返回版本或帮助信息。
- [ ] `qwenpaw init --defaults` 后存在 `working/config.json`。
- [ ] `agents/default/agent.json` 存在，并指向正确 workspace。
- [ ] `skills/research_brief/SKILL.md` 存在。
- [ ] `demo_knowledge/qwenpaw_colab_notes.md` 存在。
- [ ] Console 端口 `8088` 可访问。
- [ ] Console 能使用配置的模型完成一次回复。
- [ ] `/api/console/chat` 能完成一次测试；如果当前版本 API 变更，已记录实际 endpoint 和 payload。

## 从 PoC 到可用系统还要补什么

如果你想把这个 Colab PoC 迁到服务器或团队内部使用，至少补齐这些项：

1. **凭证治理**：API key 放环境变量或 secret manager，不写入 Notebook、Git、日志。
2. **访问控制**：公网入口必须有 TLS、认证、IP 限制或 VPN。
3. **工具权限**：默认只读；写文件、执行命令、访问数据库必须显式授权。
4. **审计日志**：记录 prompt、tool call、文件读取、外部 API 调用、错误和耗时。
5. **回滚机制**：skill、provider、agent config 改动要可 diff、可回滚。
6. **测试分层**：先本地文件读取，再模型调用，再工具调用，最后才接外部系统。
7. **资源限制**：给 API 调用设置 timeout、最大 token、最大并发和预算上限。

## 推荐的第一组实验

### 实验 1：本地知识问答

Prompt：

```text
Read demo_knowledge/qwenpaw_colab_notes.md and summarize what this workspace can do.
Separate verified facts from suggestions.
```

验收：回答必须引用 workspace notes 中的事实，不得编造未存在的文件。

### 实验 2：研究简报 skill

Prompt：

```text
Use the research_brief skill style.
Question: Should this QwenPaw workspace be used for a weekly market research assistant?
Use only local workspace notes unless you explicitly say external search is needed.
```

验收：输出包含 answer / evidence / risks / recommended next step。

### 实验 3：API 自动化回归

用第 9 步的 `stream_chat()` 固定同一个 prompt，每次修改 skill 后跑一次，检查：

- 是否能成功返回。
- 是否仍区分事实和推论。
- 是否没有泄露 access token 或 provider key。
- 是否没有建议执行危险命令。

## 结论

原文的价值不在于“又一个 Agent 工具安装教程”，而在于给出了一个完整 Agent 工作区的骨架：

```text
工作目录 + 凭证目录 + Agent 配置 + 自定义 skill + 本地知识文件 + Console + API 测试
```

这套结构可以迁移到其他 Agent 框架：先建立可审计的 workspace，再接模型，再加技能和工具，最后用 API 测试闭环。不要反过来先暴露公网服务或接生产系统。