从零构建AI文档协作工作台：Skill Runtime驱动的智能引导系统实战

---
name: "doc-coauthoring"
description: "Guide users through a structured workflow for co-authoring
documentation. Trigger when user mentions writing docs, drafting specs..."
---

# Doc Co-Authoring Workflow

## Stage 1: Context Gathering
询问文档类型、受众、目的、约束...

## Stage 2: Refinement & Structure
逐节优化，提出 5-20 个要点供用户筛选...

## Stage 3: Reader Testing
用新上下文 AI 验证文档是否清晰...

class SkillLoader:
    def __init__(self, skills_dir: Path):
        self.skills_dir = skills_dir
        self.skills: dict[str, dict] = {}
        self.load_skills()  # 启动时自动扫描

    def load_skills(self) -> None:
        """扫描 skills/ 下所有子目录"""
        for skill_dir in self.skills_dir.iterdir():
            if not skill_dir.is_dir():
                continue
            skill_md = skill_dir / "SKILL.md"
            if skill_md.exists():
                skill = self.parse_skill_md(skill_md)
                if skill:
                    self.skills[skill["name"]] = skill

def parse_skill_md(self, path: Path) -> Optional[dict]:
    content = path.read_text(encoding="utf-8")
    match = re.match(r"^---\s*\n(.*?)\n---\s*\n(.*)$", content, re.DOTALL)
    if not match:
        return None
    frontmatter, body = match.groups()
    metadata = {}
    for line in frontmatter.strip().split("\n"):
        if ":" in line:
            key, value = line.split(":", 1)
            metadata[key.strip()] = value.strip().strip("\"'")
    return {
        "name": metadata["name"],
        "description": metadata["description"],
        "body": body.strip(),
        "dir": path.parent,  # 保存目录引用，用于探测关联资源
    }

def format_skill_for_prompt(self, name: str) -> str:
    content = self.get_skill_content(name)  # body + 关联资源列表
    return f"""<skill-loaded name="{name}">
{content}
</skill-loaded>

请按照上述技能说明完成用户的任务。"""

def get_skill_content(self, name: str) -> str:
    skill = self.skills[name]
    content = f"# 技能: {skill['name']}\n\n{skill['body']}"
    # 自动探测 scripts/ references/ assets/ 下的文件
    for folder, label in [("scripts", "脚本"), ("references", "参考文档")]:
        folder_path = skill["dir"] / folder
        if folder_path.exists():
            files = list(folder_path.glob("*"))
            if files:
                content += f"\n- {label}: {', '.join(f.name for f in files)}"
    return content

_skills_loader: Optional[SkillLoader] = None

def get_skills_loader() -> SkillLoader:
    global _skills_loader
    if _skills_loader is None:
        _skills_loader = SkillLoader(Path("skills"))
    return _skills_loader

def reload_skills() -> SkillLoader:
    """运行时热重载，配合 POST /api/skills/reload"""
    global _skills_loader
    _skills_loader = None
    return get_skills_loader()

def get_skill_tool(self) -> dict:
    skill_descriptions = self._skills.get_descriptions()
    # 输出示例：
    # - **doc-coauthoring**: Guide users through...
    # - **docker**: Docker 容器化专家...
    # - **kubernetes**: K8s 部署配置专家...
    return {
        "type": "function",
        "function": {
            "name": "load_skill",
            "description": f"加载技能获取领域专业知识。\n"
                           f"可用技能：\n{skill_descriptions}",
            "parameters": {
                "type": "object",
                "properties": {
                    "skill_name": {"type": "string", "description": "技能名称"}
                },
                "required": ["skill_name"]
            }
        }
    }

def chat(self, messages, temperature=0.7, max_tokens=2000, use_tools=False):
    # 有技能时才注入 Tool 定义
    tools = [self.get_skill_tool()] if use_tools and self._skills.list_skills() else None
    
    response = self.sync_client.chat.completions.create(
        model=self.model, messages=messages, tools=tools, ...
    )
    message = response.choices[0].message
    finish_reason = response.choices[0].finish_reason
    
    # 关键分支：模型要求调用 Tool
    if finish_reason == "tool_calls" and message.tool_calls:
        return self._handle_tool_calls_sync(messages, message, ...)
    return message.content or "", None

def _handle_tool_calls_sync(self, messages, assistant_message, ...):
    # 1. 将 assistant 的 tool_call 消息加入历史
    new_messages = messages + [assistant_message]
    
    # 2. 逐个执行 tool_call
    for tool_call in assistant_message.tool_calls:
        func_name = tool_call.function.name
        func_args = json.loads(tool_call.function.arguments)
        
        if func_name == "load_skill":
            result = self.execute_skill_tool(func_args["skill_name"])
        else:
            result = f"未知工具: {func_name}"
        
        # 3. 以 tool role 追加结果（必须关联 tool_call_id）
        new_messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": result,
        })
    
    # 4. 带完整消息链重新请求（use_tools=False 防无限循环）
    return self.chat(new_messages, ..., use_tools=False)

def get_stage(self, session: Session) -> str:
    round_num = session.current_round
    max_rounds = session.max_rounds  # 默认 10
    if round_num <= max_rounds * 0.3:   return "context"      # 1-3 轮
    elif round_num <= max_rounds * 0.8: return "refinement"   # 4-8 轮
    else:                                return "testing"      # 9-10 轮

def build_messages(self, session, messages, document, user_input):
    ai_messages = [
        {"role": "system", "content": SYSTEM_PROMPT},  # 含 Skill 内容
    ]
    # 注入当前状态（阶段、轮次、完成度、文档全文）
    ai_messages.append({"role": "user", "content": f"""
**轮次**: {session.current_round}/{session.max_rounds}
**文档完成度**: {int(analysis["completeness"] * 100)}%
### 当前文档内容
```markdown
{document.content}
```"""})
    # 最近 10 条历史（滑动窗口）
    for msg in messages[-10:]:
        ai_messages.append({"role": msg.role.value, "content": msg.content})
    # 用户最新输入
    ai_messages.append({"role": "user", "content": user_input})
    return ai_messages

def analyze_content(self, content: str) -> dict:
    sections = [...]  # 检测 h1/h2/h3 章节
    has_placeholders = any(p in content for p in ["待补充", "tbd", "..."])
    completeness = 0.0
    if len(content) > 500:  completeness = 0.5
    if len(content) > 1000 and not has_placeholders: completeness = 0.7
    if len(content) > 2000 and len(sections) >= 3:   completeness = 0.9
    return {"sections": sections, "completeness": completeness, ...}

class DeploymentStructure(BaseModel):
    basic_info: ModuleStatus      # 服务名称、版本、描述
    image_config: ModuleStatus    # 镜像地址、标签、拉取策略
    resource_config: ModuleStatus # CPU/内存 requests/limits
    env_vars: ModuleStatus        # 环境变量、密钥引用
    network_config: ModuleStatus  # 端口、Service 类型、Ingress
    storage_config: ModuleStatus  # 挂载卷、持久化
    health_check: ModuleStatus    # 存活/就绪探针
    deploy_strategy: ModuleStatus # 副本数、滚动更新

@api.post("/api/sessions/stream")
async def create_session_stream(request):
    async def generate():
        async for chunk in coauthor.generate_guidance_stream(...):
            if chunk["type"] == "content":
                yield f"data: {json.dumps(chunk)}\n\n".encode("utf-8")
                await asyncio.sleep(0)  # 强制 flush
            elif chunk["type"] == "done":
                storage.save_session(session, messages, document)
                yield f"data: {json.dumps(done_data)}\n\n".encode("utf-8")
    
    return StreamingResponse(generate(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

FEATURE_AI_GUIDANCE=true      # AI 引导
FEATURE_SHOW_REASONING=true   # 推理过程展示
FEATURE_LIVE_PREVIEW=true     # 实时预览
FEATURE_EXPORT=true           # 导出功能