Plain Text
# 大模型基础配置（支持OpenAI/SiliconFlow等厂商）
API_BASE="你的模型接口地址"
API_KEY="你的模型API_KEY"
MODEL_NAME="你使用的模型ID"
MODEL_PROVIDER="模型厂商，如openai/siliconflow"
LLM_SSL_VERIFY="False" # 生产环境建议设置为True

# Agent核心配置
MAX_ITERATIONS="10" # ReAct循环最大迭代次数，避免死循环
SKILLS_DIR="./skills" # 本地Skill存放目录，生成的Skill将保存在这里
FILES_BASE_DIR="./inputs" # 输入文件目录，存放复盘文档、原始数据等
OUTPUT_DIR="./outputs" # 输出目录，存放生成的Skill、复盘报告等

# 可选：GitHub Token，用于拉取远程公共Skill，无Token会有API调用频率限制
GITHUB_TOKEN="你的GitHub Fine-grained Token"

Python
import os
import asyncio
from pathlib import Path
import requests
from dotenv import load_dotenv
import markitdown
from openjiuwen.core.common.logging import logger

# 本地文档预处理：将PDF/DOCX转换为Markdown格式
def local_file_to_markdown(file_path, files_base_dir):
    """
    将本地业务文档转换为Markdown格式，供Skills Creator解析
    :param file_path: 原始文档文件名（如"2025双十一竞品复盘标准.pdf"）
    :param files_base_dir: 输入文件根目录（对应.env中的FILES_BASE_DIR）
    :return: 转换后的Markdown文件路径
    """
    filename = Path(file_path).name
    os.makedirs(files_base_dir, exist_ok=True)
    
    source_path = Path(files_base_dir) / filename
    md_path = source_path.with_suffix(".md")

    # 调用markitdown完成文档格式转换
    md = markitdown.MarkItDown()
    result = md.convert(str(source_path))

    # 保存转换后的Markdown文件
    with md_path.open("w", encoding="utf-8") as f:
        f.write(result.text_content)

    logger.info(f"文档转换完成，Markdown已保存至: {str(md_path)}")
    return md_path

Python
from openjiuwen.dev_tools.skill_creator.skill_creator import SkillCreator

async def main():
    # 加载环境变量
    load_dotenv()
    files_base_dir = os.getenv("FILES_BASE_DIR", str(Path(__file__).resolve().parent))
    output_dir = os.getenv("OUTPUT_DIR", "")
    skills_dir = os.getenv("SKILLS_DIR", "./skills")

    # 1. 预处理业务文档，转换为Markdown格式
    review_standard_file = "2025双十一竞品复盘标准.pdf"
    md_path = local_file_to_markdown(review_standard_file, files_base_dir)

    # 2. 初始化SkillCreator实例
    skill_creator = SkillCreator()
    # 内部初始化ReActAgent，完成基础环境配置
    await skill_creator.create_agent()

    # 3. 定义生成指令，运行SkillCreator生成专属Skill
    query = f"""
    基于文件 {md_path} 生成一个电商大促竞品分析复盘Skill，要求：
    1. 严格遵循文档中的复盘SOP、数据口径和输出规范
    2. 支持自定义竞品列表、商品类目、大促时间周期
    3. 自动生成数据拉取、折扣率计算、爆款分析的对应脚本
    4. 内置标准的复盘报告输出模板
    """
    # 执行Skill生成，产物将保存至SKILLS_DIR指定目录
    res = await skill_creator.generate(query, skills_dir)
    logger.info(f"Skill生成完成，结果: {res.get('output', res)}")

if __name__ == "__main__":
    asyncio.run(main())

Plain Text
promo-review-skill/
├── SKILL.md          # 必需：Skill核心定义文件，包含元数据与执行SOP
├── scripts/          # 可选：可执行脚本，如数据拉取、指标计算脚本
│   ├── fetch_competition_data.py
│   └── promotion_indicator_calc.py
├── references/       # 可选：补充知识，如口径规范、历史复盘案例
│   ├── 大促数据口径规范.md
│   └── 2023-2025年竞品复盘优秀案例.md
└── assets/           # 可选：静态资源，如报告模板、品牌Logo
    └── 电商大促复盘报告模板.docx

Markdown
---
name: "promo-review-skill"
description: "用于生成电商大促（双十一、618等）竞品分析复盘报告。当用户需求涉及大促竞品分析、运营复盘、活动总结、折扣力度对比、爆款策略分析时触发。"
---

# 电商大促竞品复盘Skill执行手册
## 一、适用范围
本Skill适用于电商平台全类目大促活动的竞品分析与运营复盘，支持自定义竞品、类目、时间周期，严格遵循团队统一的数据口径与复盘规范。

## 二、核心执行流程
1.  **范围确认**：优先与用户明确待分析的竞品列表、商品类目、大促时间周期，未明确时使用默认值（竞品：A/B/C；类目：美妆；周期：大促全程）
2.  **数据获取**：调用`scripts/fetch_competition_data.py`脚本，传入确认后的参数，拉取竞品商品、价格、销量、营销活动等原始数据
3.  **指标计算**：调用`scripts/promotion_indicator_calc.py`脚本，按照规范计算综合折扣率、动销率、UV价值等核心指标，严格执行文档中的口径规则
4.  **多维度分析**：
    - 价格力分析：对比竞品核心商品的折扣力度、定价策略、优惠玩法
    - 爆款策略分析：提取竞品销量Top3商品的标题、卖点、赠品组合，生成卖点词云
    - 营销节奏分析：对比竞品预售、开门红、返场全周期的活动节奏与流量策略
5.  **报告生成**：基于`assets/电商大促复盘报告模板.docx`模板，填充分析数据与结论，生成完整复盘报告，输出至指定目录
6.  **结果交付**：向用户反馈复盘核心结论，同步报告文件路径，可根据用户需求调整分析维度

## 三、使用规范
1.  所有指标计算必须优先使用脚本执行，禁止擅自修改数据口径
2.  分析过程中遇到口径不明确的场景，需加载`references/大促数据口径规范.md`获取补充信息
3.  报告生成必须使用指定模板，不得擅自修改核心章节结构
4.  所有生成的文件必须保存至用户指定的输出目录，不得随意更改路径

Plain Text
import os
import asyncio
from pathlib import Path
from dotenv import load_dotenv
from openjiuwen.core.common.logging import logger
from openjiuwen.core.runner import Runner
from openjiuwen.core.skills import GitHubTree
from openjiuwen.core.sys_operation import SysOperationCard, OperationMode, LocalWorkConfig
from openjiuwen.core.single_agent import AgentCard
from openjiuwen.core.single_agent import ReActAgent, ReActAgentConfig

async def main():
    # 1. 加载环境变量
    load_dotenv()
    skills_dir = Path(os.getenv("SKILLS_DIR", "./skills")).expanduser().resolve()
    files_base_dir = os.getenv("FILES_BASE_DIR", str(Path(__file__).resolve().parent))
    output_dir = os.getenv("OUTPUT_DIR", "./outputs")
    max_iterations = int(os.getenv("MAX_ITERATIONS", 10))

    # 2. 大模型配置加载
    api_base = os.getenv("API_BASE", "")
    api_key = os.getenv("API_KEY", "")
    model_name = os.getenv("MODEL_NAME", "")
    model_provider = os.getenv("MODEL_PROVIDER", "")
    verify_ssl = os.getenv("LLM_SSL_VERIFY", "False")

    # 3. 创建ReActAgent实例，定义Agent基础身份
    agent = ReActAgent(card=AgentCard(
        name="ecommerce_promo_agent",
        description="电商大促运营专属智能体，具备竞品分析、复盘报告生成、活动数据核算能力"
    ))

    # 4. 定义系统提示词，明确文件路径与核心规则
    system_prompt = (
        "你是电商行业专业的运营分析专家，严格遵循用户所在团队的业务规范执行任务。\n"
        f"所有用户提供的业务文档、原始数据均位于目录: {files_base_dir}\n"
        f"所有生成的报告、文件、处理结果，必须全部保存至目录: {output_dir}\n"
        "执行任务时，优先使用已注册的Skill完成，不得擅自偏离Skill定义的执行流程。\n"
    )

    # 5. 配置本地系统操作权限，支持文件读写、代码执行、Shell命令运行
    sysop_card = SysOperationCard(
        mode=OperationMode.LOCAL,
        work_config=LocalWorkConfig(work_dir=None),
    )
    # 将系统操作权限注册到Runner资源管理器
    Runner.resource_mgr.add_sys_operation(sysop_card)

    # 6. 配置Agent核心参数，完成全链路初始化
    cfg = (ReActAgentConfig()
           .configure_model_client(
                provider=model_provider,
                api_key=api_key,
                api_base=api_base,
                model_name=model_name,
                verify_ssl=verify_ssl,
            )
           .configure_prompt_template([{"role": "system", "content": system_prompt}])
           .configure_max_iterations(max_iterations)
           .configure_context_engine(
                max_context_message_num=None,
                default_window_round_num=None
           )
        )
    # 绑定系统操作权限，使Agent具备本地执行能力
    cfg.sys_operation_id = sysop_card.id
    # 应用配置到Agent实例
    agent.configure(cfg)

Plain Text
    # 7. 向Agent注册Skill，支持本地注册与远程拉取两种模式
    if skills_dir.exists():
        # 可选：从GitHub远程拉取公共Skill到本地
        github_token = os.getenv("GITHUB_TOKEN", "")
        await agent.register_remote_skills(
            skills_dir=skills_dir,
            github_tree=GitHubTree(
                repo_owner="openJiuwen",
                repo_name="official-skills",
                tree_ref="HEAD",
                directory="skills/ecommerce", # 电商领域公共Skill目录
            ),
            token=github_token
        )
        # 核心：注册本地Skill目录下的所有Skill（包括我们生成的promo-review-skill）
        await agent.register_skill(str(skills_dir))
        logger.info(f"Skill注册完成，已加载目录: {str(skills_dir)}")

Plain Text
name: "promo-review-skill"
description: "用于生成电商大促（双十一、618等）竞品分析复盘报告。当用户需求涉及大促竞品分析、运营复盘、活动总结、折扣力度对比、爆款策略分析时触发。"

Python
# scripts/fetch_competition_data.py
import pandas as pd
import os
from pathlib import Path

def fetch_competition_data(competition_list, category, date_range, output_path):
    """
    拉取指定竞品、类目、时间周期的大促核心数据
    :param competition_list: 竞品列表，如["竞品A", "竞品B"]
    :param category: 商品类目，如"美妆"
    :param date_range: 时间周期，如"2025-11-01至2025-11-11"
    :param output_path: 数据保存路径
    :return: 拉取结果说明与文件路径
    """
    # 此处对接企业内部数据平台/数仓接口，拉取原始数据
    # 示例：模拟数据拉取与保存
    os.makedirs(output_path, exist_ok=True)
    file_path = Path(output_path) / f"{category}_竞品数据_{date_range}.xlsx"
    
    # 模拟数据生成（生产环境替换为真实数仓查询逻辑）
    demo_data = pd.DataFrame({
        "竞品": [],
        "商品ID": [],
        "商品名称": [],
        "吊牌价": [],
        "活动价": [],
        "券后价": [],
        "销量": [],
        "营销玩法": []
    })
    demo_data.to_excel(file_path, index=False)
    
    return f"数据拉取完成，文件已保存至: {str(file_path)}"

# 接收Agent传入的参数，执行函数并返回结果
if __name__ == "__main__":
    import sys
    competition_list = sys.argv[1].split(",")
    category = sys.argv[2]
    date_range = sys.argv[3]
    output_path = sys.argv[4]
    result = fetch_competition_data(competition_list, category, date_range, output_path)
    print(result)

Python
    # 8. 定义用户查询指令，运行Agent完成任务
    query = f"帮我生成一份2025年双十一竞品分析复盘草稿，重点看竞品A、B在美妆类目的折扣力度和爆款策略，数据周期为2025-11-01至2025-11-11。"
    # 调用Runner运行Agent，传入查询指令与会话ID
    res = await Runner.run_agent(
        agent=agent,
        inputs={"query": query, "conversation_id": "promo_review_202511"}
    )
    # 输出执行结果
    logger.info(f"任务执行完成，结果: {res.get('output', res)}")

# 完整程序入口
if __name__ == "__main__":
    asyncio.run(main())