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

/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 日志

OPENAI_API_KEY
OPENROUTER_API_KEY
DASHSCOPE_API_KEY
DEEPSEEK_API_KEY
GEMINI_API_KEY
GOOGLE_API_KEY

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)

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

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)

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

@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)

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)

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

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

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")

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)

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

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

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)

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)

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."
)

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

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

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

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.

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