diff --git a/skillagent/agent.py b/skillagent/agent.py index d39e402..6631106 100644 --- a/skillagent/agent.py +++ b/skillagent/agent.py @@ -5,6 +5,7 @@ from dataclasses import dataclass, field from pydantic import BaseModel, Field, ValidationError from typing import Literal from appPublic.worker import awaitify +from appPublic.streamhttpclient import StreamHttpClient, liner from .skillkit_wrapper import SkillkitWrapper # --------------------------- @@ -12,32 +13,32 @@ from .skillkit_wrapper import SkillkitWrapper # --------------------------- @dataclass class SkillDecision: - skill: str - params: dict - reason: Optional[str] = None + skill: str + params: dict + reason: Optional[str] = None @dataclass class PlanState: - user_intent: str - skill: str + user_intent: str + skill: str script: str - params: dict - missing: List[str] = field(default_factory=list) + params: dict + missing: List[str] = field(default_factory=list) # --------------------------- # 自定义异常 # --------------------------- class MissingParams(Exception): - def __init__(self, skill: str, fields: List[str]): - self.skill = skill - self.fields = fields + def __init__(self, skill: str, fields: List[str]): + self.skill = skill + self.fields = fields # --------------------------- # LLM 接口（可替换为你的模型） # --------------------------- class LLM: - async def complete(self, prompt: str) -> str: - raise NotImplementedError + async def complete(self, prompt: str) -> str: + raise NotImplementedError # --------------------------- # DummyLLM 示例（测试用） @@ -45,38 +46,39 @@ class LLM: class DummyLLM(LLM): def __init__(self, llmid, apikey): self.llmid = llmid - self.akikey = apikey + self.apikey = apikey - async def complete(self, prompt: str) -> str: + async def complete(self, prompt: str) -> str: hc = StreamHttpClient() - headers = { - 'Authorization': f'Bearer {self.apikey}', - 'Content-Type': 'application/json' - } + headers = { + 'Authorization': f'Bearer {self.apikey}', + 'Content-Type': 'application/json' + } d = { - 'llmid': self.llmid, - 'prompt': prompt - } + 'llmid': self.llmid, + 'prompt': prompt + } + url = 'https://opencomputing.ai/v1/llm' reco = hc('POST', url, headers=headers, data=json.dumps(d)) doc = '' - async for chunk in liner(reco): - try: - d = json.loads(chunk) - except Exception as e: - print(f'****{chunk=} error {e} {format_exc()}') - continue + async for chunk in liner(reco): + try: + d = json.loads(chunk) + except Exception as e: + print(f'****{chunk=} error {e} {format_exc()}') + continue if d.get('content'): - doc = f'{doc}{d["content"]}' - else: - print(f'{f}:{d} error') - return json.loads(doc) + doc = f'{doc}{d["content"]}' + else: + print(f'{d} error') + return doc # --------------------------- # Agent 实现 # --------------------------- class Agent: - def __init__(self, llm: LLM, skillkit): - self.llm = llm + def __init__(self, llm: LLM, skillkit): + self.llm = llm self.skillkit = skillkit self.skills = None self.loaded = False @@ -88,30 +90,30 @@ class Agent: for s in self.skills: self.skillkit.load_skill(s.name) - # --------------------------- - # plan: 多 skill 候选 + 参数抽取 - # --------------------------- - async def plan(self, user_text: str): + # --------------------------- + # plan: 多 skill 候选 + 参数抽取 + # --------------------------- + async def plan(self, user_text: str): self.load_skills() - candidates = await self._candidate_skills(user_text) - decision = await self._plan_with_candidates(user_text, candidates) - try: - validated_params = self._validate_params(decision) - except MissingParams as e: - question = await self._ask_user_for_params(user_text, decision.skill, e.fields) - state = PlanState( - user_intent=user_text, - skill=decision.skill, + candidates = await self._candidate_skills(user_text) + decision = await self._plan_with_candidates(user_text, candidates) + try: + validated_params = self._validate_params(decision) + except MissingParams as e: + question = await self._ask_user_for_params(user_text, decision.skill, e.fields) + state = PlanState( + user_intent=user_text, + skill=decision.skill, script=decision.script, - params=decision.params, - missing=e.fields - ) - return { + params=decision.params, + missing=e.fields + ) + return { "type": "clarification", "state": state, "question": question } - return { + return { "type": "script_call", "script": decision.script, "skill": decision.skill, @@ -122,15 +124,15 @@ class Agent: def get_scripts(self, skillname): return self.skillkit.get_skill_scripts(skillname) - # --------------------------- - # resume: 补 missing 参数 - # --------------------------- - async def resume(self, state: PlanState, user_reply: str): - skill_spec = next(s for s in self.skills if s.name == state.skill) - schema_fields = next(s.params for s in skill.scripts if s.name==state.script) + # --------------------------- + # resume: 补 missing 参数 + # --------------------------- + async def resume(self, state: PlanState, user_reply: str): + skill_spec = next(s for s in self.skills if s.name == state.skill) + schema_fields = self.skillkit.get_script_params(state.skill, state.script) if schema_fields is None: schema_fields = [] - prompt = f""" + prompt = f""" You are an agent helping a user fill parameters for a skill. Skill name: {state.skill} @@ -155,35 +157,38 @@ Task: - All output must match the skill parameter schema. - Output JSON only with the missing parameters. """ - raw = await self.llm.complete(prompt) - new_params = json.loads(raw) + raw = await self.llm.complete(prompt) + new_params = json.loads(raw) - state.params.update(new_params) + state.params.update(new_params) - # 校验 schema - try: - validated = self._validate_params(SkillDecision(skill=state.skill, params=state.params)) - except MissingParams as e: - state.missing = e.fields - question = await self._ask_user_for_params(state.user_intent, state.skill, e.fields) - return {"type": "clarification", "state": state, "question": question} + # 校验 schema + try: + validated = self._validate_params(SkillDecision(skill=state.skill, params=state.params)) + except MissingParams as e: + state.missing = e.fields + question = await self._ask_user_for_params(state.user_intent, state.skill, e.fields) + return {"type": "clarification", "state": state, "question": question} - # 参数完整，返回可直接调用 skill - return {"type": "skill_call", "skill": state.skill, "params": validated} + # 参数完整，返回可直接调用 skill + return {"type": "skill_call", "skill": state.skill, "params": validated} - # --------------------------- - # 内部方法 - # --------------------------- + # --------------------------- + # 内部方法 + # --------------------------- - def scripts_info(self, skill): + def scripts_info(self, skill_name): d = [] + skill = self.skillkit.load_skill(skill_name) for s in skill.scripts: - d.append( f'name:{s.name}, description:{s.description}, params:{str(s.params}' - return "Scripts: '::'.join(d) + params = self.skillkit.get_script_params(skill_name, s.name) + print(f'{params=}') + d.append( f'name:{s.name}, description:{s.description}, params:{str(params)}') + return "Scripts: " + '::'.join(d) - async def _candidate_skills(self, user_text: str): - skill_list = "\n".join(f"- skillname:{s.name}({s.description}): {self.scripts_info(s)}" for s in self.skills) - prompt = f""" + async def _candidate_skills(self, user_text: str): + skill_list = "\n".join(f"- skillname:{s.name}({s.description}): {self.scripts_info(s.name)}" for s in self.skills) + prompt = f""" User request: \"\"\"{user_text}\"\"\" @@ -195,15 +200,15 @@ Select up to 3 most relevant skill's scripts. Output JSON list only. """ - raw = await self.llm.complete(prompt) - return json.loads(raw) + raw = await self.llm.complete(prompt) + return json.loads(raw) - async def _plan_with_candidates(self, user_text: str, candidates: list[str]): - specs = [s for s in self.skills if s.name in candidates] - spec_desc = "\n".join( - f"- {s.name}: inputs={list(s.schema.model_fields.keys())}" for s in specs if s.schema - ) - prompt = f""" + async def _plan_with_candidates(self, user_text: str, candidates: list[str]): + specs = [s for s in self.skills if s.name in candidates] + spec_desc = "\n".join( + f"- {s.name}: inputs={list(s.schema.model_fields.keys())}" for s in specs if s.schema + ) + prompt = f""" User request: \"\"\"{user_text}\"\"\" @@ -226,23 +231,23 @@ Output: "reason": "..." }} """ - raw = await self.llm.complete(prompt) - return SkillDecision(**json.loads(raw)) + raw = await self.llm.complete(prompt) + return SkillDecision(**json.loads(raw)) - def _validate_params(self, decision: SkillDecision): - spec = next(s for s in self.skills if s.name == decision.skill) - if not spec.schema: - return decision.params - try: - return spec.schema(**decision.params).dict() - except ValidationError as e: - missing = [err["loc"][0] for err in e.errors() if err["type"] == "missing"] - if missing: - raise MissingParams(decision.skill, missing) - raise + def _validate_params(self, decision: SkillDecision): + spec = next(s for s in self.skills if s.name == decision.skill) + if not spec.schema: + return decision.params + try: + return spec.schema(**decision.params).dict() + except ValidationError as e: + missing = [err["loc"][0] for err in e.errors() if err["type"] == "missing"] + if missing: + raise MissingParams(decision.skill, missing) + raise - async def _ask_user_for_params(self, user_text: str, skill: str, fields: List[str]): - prompt = f""" + async def _ask_user_for_params(self, user_text: str, skill: str, fields: List[str]): + prompt = f""" User request: \"\"\"{user_text}\"\"\" @@ -251,15 +256,15 @@ The script "{script} in skill "{skill}" requires the following missing parameter Ask the user a concise clarification question. """ - return await self.llm.complete(prompt) + return await self.llm.complete(prompt) # --------------------------- # 测试运行 # --------------------------- -async def skillagent(llm, apikey, user_skillroot, sys_skillroot): - llm = DummyLLM('8L4hFJ4QpSMyu1UP03Juo', 'eYgNuD6sVQgbj-khOOUNU') - skillkit = SkillKitWrapper(skill_rootpath) - agent = Agent(llm, skillkit) +async def skillagent(llm, apikey, user_skillroot, sys_skillroot=None): + llm = DummyLLM('8L4hFJ4QpSMyu1UP03Juo', 'eYgNuD6sVQgbj-khOOUNU') + skillkit = SkillkitWrapper(user_skillroot) + agent = Agent(llm, skillkit) while True: print('What you want to do?') diff --git a/skillagent/skillkit_wrapper.py b/skillagent/skillkit_wrapper.py index 7935464..6f9536e 100644 --- a/skillagent/skillkit_wrapper.py +++ b/skillagent/skillkit_wrapper.py @@ -3,6 +3,7 @@ from skillkit import SkillManager import yaml from pathlib import Path from typing import Dict, Any +from appPublic.dictObject import DictObject def find_missing_params( input_schema: Dict[str, Any], @@ -19,7 +20,7 @@ def find_missing_params( return missing -def load_schemas(yaml_path: str) -> Dict[str, Any]: +def load_schemas(path) -> Dict[str, Any]: """ 从 YAML 文件中读取 script 输入参数定义 @@ -32,47 +33,46 @@ def load_schemas(yaml_path: str) -> Dict[str, Any]: } } """ - path = Path(yaml_path) if not path.exists(): raise FileNotFoundError(f"Script yaml not found: {yaml_path}") with path.open("r", encoding="utf-8") as f: data = yaml.safe_load(f) - - if "script" not in data or "inputs" not in data: - raise ValueError("Invalid script yaml format") - - return { - "script": data["script"], - "description": data.get("description", ""), - "inputs": data["inputs"], - } + + return DictObject(data) class SkillkitWrapper: def __init__(self, user_skillsroot, sys_skillsroot=None): - self.client = SkillManager(project_skill_dir=skillroot, + self.client = SkillManager(project_skill_dir=user_skillsroot, anthropic_config_dir=sys_skillsroot) self.client.discover() + self.schemas = {} def list_skills(self): return self.client.list_skills() - def load_skill(self, skillname): + def load_skill(self, skill_name): skill = self.client.load_skill(skill_name) - if not hasattr(skill, 'schemas'): - fp = os.path.join(skill.base_dir, 'schemas.yaml') - if os.path.exists(fp): - data = load_schema(fp) - skill.schemas = data - for s in skill.scripts: - s.params = next(sch.inputs for sch in skill.schemas if sch.script==script_name) + print(skill, dir(skill)) + schemaspath = skill.base_directory / 'schemas.yaml' + if schemaspath.exists(): + if not self.schemas.get(skill_name): + data = load_schemas(schemaspath) + self.schemas[skill_name] = data + print(f'{data=}, {str(schemaspath)}') return skill def get_script_params(self, skill_name, script_name): skill = self.load_skill(skill_name) - return next(s.params for s in skill.scripts if s.name==script_name) + d = self.schemas.get('skill_name') + if not d: + return [] + m = d.scripts.get(script_name) + if not m: + return [] + return m.inputs def get_skill_scripts(self, skill_name): skill = self.load_skill(skill_name)