yumoqing/skillagent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfix

Browse Source
This commit is contained in:
yumoqing 2026-01-20 18:27:11 +08:00
parent 7fbf7f7a31
commit 6dda573aed
2 changed files with 133 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

219
skillagent/agent.py
View File

@ -5,6 +5,7 @@ from dataclasses import dataclass, field
from pydantic import BaseModel, Field, ValidationError from pydantic import BaseModel, Field, ValidationError
from typing import Literal from typing import Literal
from appPublic.worker import awaitify from appPublic.worker import awaitify
from appPublic.streamhttpclient import StreamHttpClient, liner
from .skillkit_wrapper import SkillkitWrapper from .skillkit_wrapper import SkillkitWrapper
# --------------------------- # ---------------------------
@ -12,32 +13,32 @@ from .skillkit_wrapper import SkillkitWrapper
# --------------------------- # ---------------------------
@dataclass @dataclass
class SkillDecision: class SkillDecision:
skill: str skill: str
params: dict params: dict
reason: Optional[str] = None reason: Optional[str] = None
@dataclass @dataclass
class PlanState: class PlanState:
user_intent: str user_intent: str
skill: str skill: str
script: str script: str
params: dict params: dict
missing: List[str] = field(default_factory=list) missing: List[str] = field(default_factory=list)
# --------------------------- # ---------------------------
# 自定义异常 # 自定义异常
# --------------------------- # ---------------------------
class MissingParams(Exception): class MissingParams(Exception):
def __init__(self, skill: str, fields: List[str]): def __init__(self, skill: str, fields: List[str]):
self.skill = skill self.skill = skill
self.fields = fields self.fields = fields
# --------------------------- # ---------------------------
# LLM 接口（可替换为你的模型） # LLM 接口（可替换为你的模型）
# --------------------------- # ---------------------------
class LLM: class LLM:
async def complete(self, prompt: str) -> str: async def complete(self, prompt: str) -> str:
raise NotImplementedError raise NotImplementedError
# --------------------------- # ---------------------------
# DummyLLM 示例（测试用） # DummyLLM 示例（测试用）
@ -45,38 +46,39 @@ class LLM:
class DummyLLM(LLM): class DummyLLM(LLM):
def __init__(self, llmid, apikey): def __init__(self, llmid, apikey):
self.llmid = llmid 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() hc = StreamHttpClient()
headers = { headers = {
'Authorization': f'Bearer {self.apikey}', 'Authorization': f'Bearer {self.apikey}',
'Content-Type': 'application/json' 'Content-Type': 'application/json'
} }
d = { d = {
'llmid': self.llmid, 'llmid': self.llmid,
'prompt': prompt 'prompt': prompt
} }
url = 'https://opencomputing.ai/v1/llm'
reco = hc('POST', url, headers=headers, data=json.dumps(d)) reco = hc('POST', url, headers=headers, data=json.dumps(d))
doc = '' doc = ''
async for chunk in liner(reco): async for chunk in liner(reco):
try: try:
d = json.loads(chunk) d = json.loads(chunk)
except Exception as e: except Exception as e:
print(f'****{chunk=} error {e} {format_exc()}') print(f'****{chunk=} error {e} {format_exc()}')
continue continue
if d.get('content'): if d.get('content'):
doc = f'{doc}{d["content"]}' doc = f'{doc}{d["content"]}'
else: else:
print(f'{f}:{d} error') print(f'{d} error')
return json.loads(doc) return doc
# --------------------------- # ---------------------------
# Agent 实现 # Agent 实现
# --------------------------- # ---------------------------
class Agent: class Agent:
def __init__(self, llm: LLM, skillkit): def __init__(self, llm: LLM, skillkit):
self.llm = llm self.llm = llm
self.skillkit = skillkit self.skillkit = skillkit
self.skills = None self.skills = None
self.loaded = False self.loaded = False
@ -88,30 +90,30 @@ class Agent:
for s in self.skills: for s in self.skills:
self.skillkit.load_skill(s.name) self.skillkit.load_skill(s.name)
# --------------------------- # ---------------------------
# plan: 多 skill 候选 + 参数抽取 # plan: 多 skill 候选 + 参数抽取
# --------------------------- # ---------------------------
async def plan(self, user_text: str): async def plan(self, user_text: str):
self.load_skills() self.load_skills()
candidates = await self._candidate_skills(user_text) candidates = await self._candidate_skills(user_text)
decision = await self._plan_with_candidates(user_text, candidates) decision = await self._plan_with_candidates(user_text, candidates)
try: try:
validated_params = self._validate_params(decision) validated_params = self._validate_params(decision)
except MissingParams as e: except MissingParams as e:
question = await self._ask_user_for_params(user_text, decision.skill, e.fields) question = await self._ask_user_for_params(user_text, decision.skill, e.fields)
state = PlanState( state = PlanState(
user_intent=user_text, user_intent=user_text,
skill=decision.skill, skill=decision.skill,
script=decision.script, script=decision.script,
params=decision.params, params=decision.params,
missing=e.fields missing=e.fields
) )
return { return {
"type": "clarification", "type": "clarification",
"state": state, "state": state,
"question": question "question": question
} }
return { return {
"type": "script_call", "type": "script_call",
"script": decision.script, "script": decision.script,
"skill": decision.skill, "skill": decision.skill,
@ -122,15 +124,15 @@ class Agent:
def get_scripts(self, skillname): def get_scripts(self, skillname):
return self.skillkit.get_skill_scripts(skillname) return self.skillkit.get_skill_scripts(skillname)
# --------------------------- # ---------------------------
# resume: 补 missing 参数 # resume: 补 missing 参数
# --------------------------- # ---------------------------
async def resume(self, state: PlanState, user_reply: str): async def resume(self, state: PlanState, user_reply: str):
skill_spec = next(s for s in self.skills if s.name == state.skill) 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) schema_fields = self.skillkit.get_script_params(state.skill, state.script)
if schema_fields is None: if schema_fields is None:
schema_fields = [] schema_fields = []
prompt = f""" prompt = f"""
You are an agent helping a user fill parameters for a skill. You are an agent helping a user fill parameters for a skill.
Skill name: {state.skill} Skill name: {state.skill}
@ -155,35 +157,38 @@ Task:
- All output must match the skill parameter schema. - All output must match the skill parameter schema.
- Output JSON only with the missing parameters. - Output JSON only with the missing parameters.
""" """
raw = await self.llm.complete(prompt) raw = await self.llm.complete(prompt)
new_params = json.loads(raw) new_params = json.loads(raw)
state.params.update(new_params) state.params.update(new_params)
# 校验 schema # 校验 schema
try: try:
validated = self._validate_params(SkillDecision(skill=state.skill, params=state.params)) validated = self._validate_params(SkillDecision(skill=state.skill, params=state.params))
except MissingParams as e: except MissingParams as e:
state.missing = e.fields state.missing = e.fields
question = await self._ask_user_for_params(state.user_intent, state.skill, e.fields) question = await self._ask_user_for_params(state.user_intent, state.skill, e.fields)
return {"type": "clarification", "state": state, "question": question} return {"type": "clarification", "state": state, "question": question}
# 参数完整，返回可直接调用 skill # 参数完整，返回可直接调用 skill
return {"type": "skill_call", "skill": state.skill, "params": validated} return {"type": "skill_call", "skill": state.skill, "params": validated}
# --------------------------- # ---------------------------
# 内部方法 # 内部方法
# --------------------------- # ---------------------------
def scripts_info(self, skill): def scripts_info(self, skill_name):
d = [] d = []
skill = self.skillkit.load_skill(skill_name)
for s in skill.scripts: for s in skill.scripts:
d.append( f'name:{s.name}, description:{s.description}, params:{str(s.params}' params = self.skillkit.get_script_params(skill_name, s.name)
return "Scripts: '::'.join(d) 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): 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) skill_list = "\n".join(f"- skillname:{s.name}({s.description}): {self.scripts_info(s.name)}" for s in self.skills)
prompt = f""" prompt = f"""
User request: User request:
\"\"\"{user_text}\"\"\" \"\"\"{user_text}\"\"\"
@ -195,15 +200,15 @@ Select up to 3 most relevant skill's scripts.
Output JSON list only. Output JSON list only.
""" """
raw = await self.llm.complete(prompt) raw = await self.llm.complete(prompt)
return json.loads(raw) return json.loads(raw)
async def _plan_with_candidates(self, user_text: str, candidates: list[str]): async def _plan_with_candidates(self, user_text: str, candidates: list[str]):
specs = [s for s in self.skills if s.name in candidates] specs = [s for s in self.skills if s.name in candidates]
spec_desc = "\n".join( spec_desc = "\n".join(
f"- {s.name}: inputs={list(s.schema.model_fields.keys())}" for s in specs if s.schema f"- {s.name}: inputs={list(s.schema.model_fields.keys())}" for s in specs if s.schema
) )
prompt = f""" prompt = f"""
User request: User request:
\"\"\"{user_text}\"\"\" \"\"\"{user_text}\"\"\"
@ -226,23 +231,23 @@ Output:
"reason": "..." "reason": "..."
}} }}
""" """
raw = await self.llm.complete(prompt) raw = await self.llm.complete(prompt)
return SkillDecision(**json.loads(raw)) return SkillDecision(**json.loads(raw))
def _validate_params(self, decision: SkillDecision): def _validate_params(self, decision: SkillDecision):
spec = next(s for s in self.skills if s.name == decision.skill) spec = next(s for s in self.skills if s.name == decision.skill)
if not spec.schema: if not spec.schema:
return decision.params return decision.params
try: try:
return spec.schema(**decision.params).dict() return spec.schema(**decision.params).dict()
except ValidationError as e: except ValidationError as e:
missing = [err["loc"][0] for err in e.errors() if err["type"] == "missing"] missing = [err["loc"][0] for err in e.errors() if err["type"] == "missing"]
if missing: if missing:
raise MissingParams(decision.skill, missing) raise MissingParams(decision.skill, missing)
raise raise
async def _ask_user_for_params(self, user_text: str, skill: str, fields: List[str]): async def _ask_user_for_params(self, user_text: str, skill: str, fields: List[str]):
prompt = f""" prompt = f"""
User request: User request:
\"\"\"{user_text}\"\"\" \"\"\"{user_text}\"\"\"
@ -251,15 +256,15 @@ The script "{script} in skill "{skill}" requires the following missing parameter
Ask the user a concise clarification question. 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): async def skillagent(llm, apikey, user_skillroot, sys_skillroot=None):
llm = DummyLLM('8L4hFJ4QpSMyu1UP03Juo', 'eYgNuD6sVQgbj-khOOUNU') llm = DummyLLM('8L4hFJ4QpSMyu1UP03Juo', 'eYgNuD6sVQgbj-khOOUNU')
skillkit = SkillKitWrapper(skill_rootpath) skillkit = SkillkitWrapper(user_skillroot)
agent = Agent(llm, skillkit) agent = Agent(llm, skillkit)
while True: while True:
print('What you want to do?') print('What you want to do?')

40
skillagent/skillkit_wrapper.py
View File

@ -3,6 +3,7 @@ from skillkit import SkillManager
import yaml import yaml
from pathlib import Path from pathlib import Path
from typing import Dict, Any from typing import Dict, Any
from appPublic.dictObject import DictObject
def find_missing_params( def find_missing_params(
input_schema: Dict[str, Any], input_schema: Dict[str, Any],
@ -19,7 +20,7 @@ def find_missing_params(
return missing return missing
def load_schemas(yaml_path: str) -> Dict[str, Any]: def load_schemas(path) -> Dict[str, Any]:
""" """
从 YAML 文件中读取 script 输入参数定义 从 YAML 文件中读取 script 输入参数定义
@ -32,47 +33,46 @@ def load_schemas(yaml_path: str) -> Dict[str, Any]:
} }
} }
""" """
path = Path(yaml_path)
if not path.exists(): if not path.exists():
raise FileNotFoundError(f"Script yaml not found: {yaml_path}") raise FileNotFoundError(f"Script yaml not found: {yaml_path}")
with path.open("r", encoding="utf-8") as f: with path.open("r", encoding="utf-8") as f:
data = yaml.safe_load(f) data = yaml.safe_load(f)
if "script" not in data or "inputs" not in data: return DictObject(data)
raise ValueError("Invalid script yaml format")
return {
"script": data["script"],
"description": data.get("description", ""),
"inputs": data["inputs"],
}
class SkillkitWrapper: class SkillkitWrapper:
def __init__(self, user_skillsroot, sys_skillsroot=None): 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) anthropic_config_dir=sys_skillsroot)
self.client.discover() self.client.discover()
self.schemas = {}
def list_skills(self): def list_skills(self):
return self.client.list_skills() return self.client.list_skills()
def load_skill(self, skillname): def load_skill(self, skill_name):
skill = self.client.load_skill(skill_name) skill = self.client.load_skill(skill_name)
if not hasattr(skill, 'schemas'): print(skill, dir(skill))
fp = os.path.join(skill.base_dir, 'schemas.yaml') schemaspath = skill.base_directory / 'schemas.yaml'
if os.path.exists(fp): if schemaspath.exists():
data = load_schema(fp) if not self.schemas.get(skill_name):
skill.schemas = data data = load_schemas(schemaspath)
for s in skill.scripts: self.schemas[skill_name] = data
s.params = next(sch.inputs for sch in skill.schemas if sch.script==script_name) print(f'{data=}, {str(schemaspath)}')
return skill return skill
def get_script_params(self, skill_name, script_name): def get_script_params(self, skill_name, script_name):
skill = self.load_skill(skill_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): def get_skill_scripts(self, skill_name):
skill = self.load_skill(skill_name) skill = self.load_skill(skill_name)