AI Agent Skill工程应用实践：Skill全流程设计与调度优化，助力复杂业务场景高效落地21.9

import requests

# 原生裸工具：无校验、无兜底、不可复用
def raw_search(keyword: str) -> dict:
    return requests.get(f"https://api.search.com?q={keyword}").json()

# 工程化标准Skill（全文统一模板）
class StandardSearchSkill:
    # 固定标准化出参：code/data/msg
    def execute(self, keyword: str) -> dict:
        # 1. 统一参数校验
        if not keyword or len(keyword) > 100:
            return {"code": 400, "data": None, "msg": "关键词非法"}
        try:
            # 2. 核心能力执行
            resp = requests.get(f"https://api.search.com?q={keyword}", timeout=5)
            resp.raise_for_status()
            return {"code": 200, "data": resp.json(), "msg": "执行成功"}
        # 3. 统一异常兜底
        except Exception as e:
            return {"code": 500, "data": None, "msg": f"调用异常：{str(e)}"}

# 统一入口调用
if __name__ == "__main__":
    skill = StandardSearchSkill()
    print(skill.execute("AI Agent Skill工程"))

import time
import logging

# 全局统一日志配置（所有技能共用）
logging.basicConfig(
    filename="skill_global.log",
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    encoding="utf-8"
)

# 全局技能统一基类：封装通用日志能力
class BaseSkill:
    skill_name = "base_skill"

    def log(self, params: dict, result: dict, cost: float):
        """统一日志打印格式，全局标准化"""
        log_info = f"技能[{self.skill_name}] 入参:{params} 出参:{result} 耗时:{cost:.2f}s"
        logging.info(log_info)

# 统一调用入口
if __name__ == "__main__":
    base = BaseSkill()
    start = time.time()
    base.log({"text": "测试文本"}, {"code": 200, "msg": "成功"}, time.time() - start)

from typing import Dict

# 1. 标准化原子技能（统一模板）
class SearchAtomSkill(BaseSkill):
    skill_name = "搜索原子技能"
    def execute(self, keyword: str) -> Dict:
        return {"code": 200, "data": f"{keyword}搜索数据", "msg": "成功"}

class SummaryAtomSkill(BaseSkill):
    skill_name = "总结原子技能"
    def execute(self, content: str) -> str:
        return f"内容总结：{content[:20]}"

# 2. 复合技能：业务场景组装
class WeeklyReportSkill(BaseSkill):
    skill_name = "周报生成复合技能"

    def __init__(self):
        # 注入标准化原子技能
        self.search = SearchAtomSkill()
        self.summary = SummaryAtomSkill()

    def execute(self, query: str) -> Dict:
        # 串行编排
        search_res = self.search.execute(query)
        summary_res = self.summary.execute(search_res["data"])
        # 统一日志记录
        self.log({"query": query}, {"result": summary_res}, 0.01)
        return {"code": 200, "data": summary_res, "msg": "周报生成完成"}

# 统一调用入口
if __name__ == "__main__":
    report_skill = WeeklyReportSkill()
    print(report_skill.execute("本周工作复盘"))

import re

# 继承全局统一技能基类，完全复用日志、规范能力
class TextCleanSkill(BaseSkill):
    skill_name = "通用文本清洗技能"
    skill_version = "1.0.0"

    def execute(self, text: str) -> dict:
        # 1. 统一参数校验
        if not isinstance(text, str) or not text.strip():
            return {"code": 400, "data": "", "msg": "文本内容不能为空"}
        
        try:
            # 2. 核心业务逻辑
            clean_text = text.strip()
            clean_text = re.sub(r"\s+", " ", clean_text)
            clean_text = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9\s，。、；：]", "", clean_text)
            
            # 3. 统一日志记录
            self.log({"raw_text": text}, {"clean_text": clean_text}, 0.02)
            return {"code": 200, "data": clean_text, "msg": "清洗成功"}
        
        # 4. 统一异常兜底
        except Exception as e:
            return {"code": 500, "data": text, "msg": f"清洗异常：{str(e)}"}

# 统一测试入口
if __name__ == "__main__":
    skill = TextCleanSkill()
    print(skill.execute("  测试@@文本！！多余  空格  "))

from functools import lru_cache
import time

# 全局通用：缓存+频次管控工具类（统一复用）
class SkillLimitCache:
    def __init__(self, max_freq: int = 10):
        self.max_freq = max_freq
        self.call_cnt = 0
        self.last_reset_time = time.time()

    # 统一频次限流逻辑
    def freq_limit_check(self) -> bool:
        now = time.time()
        if now - self.last_reset_time >= 1:
            self.call_cnt = 0
            self.last_reset_time = now
        if self.call_cnt >= self.max_freq:
            return False
        self.call_cnt += 1
        return True

    # 统一内存缓存装饰器
    @staticmethod
    @lru_cache(maxsize=128)
    def cache_result(func, *args):
        return func(*args)

# 业务技能集成管控能力
class CalcSkill(BaseSkill):
    skill_name = "带缓存计算技能"
    def __init__(self):
        self.limit_cache = SkillLimitCache()

    def execute(self, num: int) -> dict:
        # 频次校验前置
        if not self.limit_cache.freq_limit_check():
            return {"code": 429, "data": None, "msg": "调用频次超限"}
        
        # 缓存复用结果
        res = self.limit_cache.cache_result(lambda x: x * 2 + 1, num)
        self.log({"num": num}, {"result": res}, 0.01)
        return {"code": 200, "data": res, "msg": "计算成功"}

# 统一调用入口
if __name__ == "__main__":
    skill = CalcSkill()
    print(skill.execute(10))

# 标准化串行工作流（统一编排模板）
class DocParseSkill(BaseSkill):
    skill_name = "文档解析技能"
    def execute(self, path: str) -> dict:
        return {"code": 200, "data": "原始文档内容", "msg": "解析成功"}

class DocCleanSkill(BaseSkill):
    skill_name = "文档清洗技能"
    def execute(self, text: str) -> dict:
        return {"code": 200, "data": "标准化清洗内容", "msg": "清洗成功"}

class DocSummarySkill(BaseSkill):
    skill_name = "文档总结技能"
    def execute(self, text: str) -> dict:
        return {"code": 200, "data": "文档核心总结", "msg": "总结成功"}

# 串行编排工作流：统一流程模板
class DocWorkflow(BaseSkill):
    skill_name = "文档处理工作流"
    def __init__(self):
        self.parse = DocParseSkill()
        self.clean = DocCleanSkill()
        self.summary = DocSummarySkill()

    def run(self, file_path: str) -> dict:
        # 固定串行流程：解析->清洗->总结
        parse_res = self.parse.execute(file_path)
        clean_res = self.clean.execute(parse_res["data"])
        summary_res = self.summary.execute(clean_res["data"])
        
        self.log({"file_path": file_path}, summary_res, 0.05)
        return summary_res

# 统一调用入口
if __name__ == "__main__":
    workflow = DocWorkflow()
    print(workflow.run("/data/test.docx"))

# 统一循环+条件分支编排模板
class BatchTextCleanSkill(BaseSkill):
    skill_name = "批量文本清洗技能"

    def single_clean(self, text: str) -> str:
        """单条文本标准化清洗"""
        return text.strip().replace("\n", "")

    def execute(self, text_list: list) -> dict:
        """批量编排：循环遍历+条件判断"""
        result = []
        for text in text_list:
            # 条件分支：非空清洗、空值跳过
            if text.strip():
                result.append(self.single_clean(text))
            else:
                result.append("空文本跳过")
        
        self.log({"batch_num": len(text_list)}, {"result": result}, 0.02)
        return {"code": 200, "data": result, "msg": "批量处理完成"}

# 统一调用入口
if __name__ == "__main__":
    batch_skill = BatchTextCleanSkill()
    test_data = ["  文本1  ", "", "  文本2\n换行"]
    print(batch_skill.execute(test_data))

import random

# 全局统一工作流上下文
class WorkContext:
    def __init__(self):
        self.params: dict = {}
        self.results: dict = {}
        self.status: str = "running"

# 统一异常重试编排模板
class RetryWorkflowSkill(BaseSkill):
    skill_name = "重试容错工作流"
    MAX_RETRY = 2

    def __init__(self, context: WorkContext):
        self.ctx = context

    def unstable_task(self) -> str:
        # 模拟不稳定接口
        if random.random() > 0.3:
            return "任务执行成功"
        raise ConnectionError("接口临时异常")

    def execute(self) -> dict:
        # 标准化重试逻辑
        for retry in range(self.MAX_RETRY + 1):
            try:
                res = self.unstable_task()
                self.ctx.status = "success"
                self.ctx.results["data"] = res
                self.log({"重试次数": retry}, {"res": res}, 0.1)
                return {"code": 200, "data": res, "msg": "执行成功"}
            except Exception as e:
                # 最终重试失败，统一兜底
                if retry == self.MAX_RETRY:
                    self.ctx.status = "failed"
                    self.ctx.results["error"] = str(e)
                    return {"code": 500, "data": None, "msg": "重试失败，任务兜底终止"}

# 统一调用入口
if __name__ == "__main__":
    ctx = WorkContext()
    workflow = RetryWorkflowSkill(ctx)
    print(workflow.execute())

# 统一标准化技能调度器
class SkillScheduler(BaseSkill):
    skill_name = "全局技能调度器"
    # 统一路由映射表
    ROUTE_MAP = {
        "总结": "summary_skill",
        "清洗": "clean_skill",
        "翻译": "translate_skill",
        "解析": "parse_skill"
    }

    def route(self, user_query: str) -> str:
        """标准化关键词路由"""
        for key, skill_id in self.ROUTE_MAP.items():
            if key in user_query:
                return skill_id
        return "default_skill"

    def execute(self, user_query: str) -> dict:
        """统一调度入口"""
        target_skill = self.route(user_query)
        self.log({"user_query": user_query}, {"target_skill": target_skill}, 0.01)
        return {"code": 200, "data": target_skill, "msg": "技能调度匹配成功"}

# 统一调用入口
if __name__ == "__main__":
    scheduler = SkillScheduler()
    print(scheduler.execute("帮我总结文档内容"))
    print(scheduler.execute("清洗这段杂乱文本"))

import time
from queue import PriorityQueue

# 统一优先级调度模板
class PriorityScheduler(BaseSkill):
    skill_name = "优先级任务调度器"

    def __init__(self):
        self.task_queue = PriorityQueue()

    def add_task(self, priority: int, name: str, timeout: int = 3):
        """添加任务：1高优/2中优/3低优"""
        self.task_queue.put((priority, name, timeout))

    def execute(self) -> dict:
        """统一调度执行+超时管控"""
        result_list = []
        while not self.task_queue.empty():
            priority, name, timeout = self.task_queue.get()
            start = time.time()
            try:
                time.sleep(1) # 模拟任务耗时
                if time.time() - start > timeout:
                    raise TimeoutError("任务超时")
                result_list.append(f"【成功】优先级{priority}-{name}")
            except Exception as e:
                result_list.append(f"【失败】优先级{priority}-{name}：{str(e)}")
        
        self.log({"任务总数": len(result_list)}, {"结果": result_list}, 0.1)
        return {"code": 200, "data": result_list, "msg": "调度完成"}

# 统一调用入口
if __name__ == "__main__":
    scheduler = PriorityScheduler()
    scheduler.add_task(3, "批量数据统计")
    scheduler.add_task(1, "核心舆情分析")
    scheduler.add_task(2, "日常文档整理")
    print(scheduler.execute())

import time
from functools import lru_cache

# 统一多级缓存优化模板
class CacheOptSkill(BaseSkill):
    skill_name = "缓存优化技能"

    # 一级缓存：全局内存缓存
    @lru_cache(maxsize=256)
    def _static_process(self, text: str) -> str:
        time.sleep(0.5) # 模拟推理耗时
        return f"标准化结果：{text[:10]}"

    # 二级缓存：自定义本地缓存
    def __init__(self):
        self.local_cache = {}

    def execute(self, cache_key: str) -> dict:
        # 缓存优先策略
        if cache_key in self.local_cache:
            res = self.local_cache[cache_key]
            self.log({"cache_key": cache_key}, {"来源": "本地缓存", "res": res}, 0.001)
            return {"code": 200, "data": res, "msg": "缓存复用成功"}
        
        # 无缓存则执行并存储
        res = self._static_process(cache_key)
        self.local_cache[cache_key] = res
        self.log({"cache_key": cache_key}, {"来源": "新执行", "res": res}, 0.5)
        return {"code": 200, "data": res, "msg": "执行成功，已缓存"}

# 统一调用入口
if __name__ == "__main__":
    skill = CacheOptSkill()
    # 首次执行
    t1 = time.time()
    skill.execute("测试缓存文本")
    print(f"首次耗时：{time.time()-t1:.2f}s")
    # 二次缓存复用
    t2 = time.time()
    skill.execute("测试缓存文本")
    print(f"缓存耗时：{time.time()-t2:.2f}s")