fixbug: 修复通用智能体role及其相关的TalkAction和SkillAction

This commit is contained in:
莘权 马 2023-12-25 22:39:03 +08:00
parent 6261251279
commit 0fdb552468
12 changed files with 541 additions and 512 deletions

View file

@ -9,6 +9,7 @@
from pathlib import Path
from typing import Dict, List, Optional
import aiofiles
import yaml
from pydantic import BaseModel, Field
@ -63,61 +64,37 @@ class SkillsDeclaration(BaseModel):
entities: Dict[str, Entity]
components: Components = None
class SkillLoader:
def __init__(self, skill_yaml_file_name: Path = None):
@staticmethod
async def load(skill_yaml_file_name: Path = None) -> "SkillsDeclaration":
if not skill_yaml_file_name:
skill_yaml_file_name = Path(__file__).parent.parent.parent / ".well-known/skills.yaml"
with open(str(skill_yaml_file_name), "r") as file:
skills = yaml.safe_load(file)
self._skills = SkillsDeclaration(**skills)
async with aiofiles.open(str(skill_yaml_file_name), mode="r") as reader:
data = await reader.read(-1)
skill_data = yaml.safe_load(data)
return SkillsDeclaration(**skill_data)
def get_skill_list(self, entity_name: str = "Assistant") -> Dict:
"""Return the skill name based on the skill description."""
entity = self.get_entity(entity_name)
entity = self.entities.get(entity_name)
if not entity:
return {}
# List of skills that the agent chooses to activate.
agent_skills = CONFIG.agent_skills
if not agent_skills:
return {}
class AgentSkill(BaseModel):
class _AgentSkill(BaseModel):
name: str
names = [AgentSkill(**i).name for i in agent_skills]
description_to_name_mappings = {}
for s in entity.skills:
if s.name not in names:
continue
description_to_name_mappings[s.description] = s.name
return description_to_name_mappings
names = [_AgentSkill(**i).name for i in agent_skills]
return {s.description: s.name for s in entity.skills if s.name in names}
def get_skill(self, name, entity_name: str = "Assistant") -> Skill:
"""Return a skill by name."""
entity = self.get_entity(entity_name)
entity = self.entities.get(entity_name)
if not entity:
return None
for sk in entity.skills:
if sk.name == name:
return sk
def get_entity(self, name) -> Entity:
"""Return a list of skills for the entity."""
if not self._skills:
return None
return self._skills.entities.get(name)
if __name__ == "__main__":
CONFIG.agent_skills = [
{"id": 1, "name": "text_to_speech", "type": "builtin", "config": {}, "enabled": True},
{"id": 2, "name": "text_to_image", "type": "builtin", "config": {}, "enabled": True},
{"id": 3, "name": "ai_call", "type": "builtin", "config": {}, "enabled": True},
{"id": 3, "name": "data_analysis", "type": "builtin", "config": {}, "enabled": True},
{"id": 5, "name": "crawler", "type": "builtin", "config": {"engine": "ddg"}, "enabled": True},
{"id": 6, "name": "knowledge", "type": "builtin", "config": {}, "enabled": True},
]
loader = SkillLoader()
print(loader.get_skill_list())

View file

@ -27,7 +27,7 @@ async def text_to_image(text, size_type: str = "512x512", openai_api_key="", mod
if CONFIG.METAGPT_TEXT_TO_IMAGE_MODEL_URL or model_url:
base64_data = await oas3_metagpt_text_to_image(text, size_type, model_url)
elif CONFIG.OPENAI_API_KEY or openai_api_key:
base64_data = await oas3_openai_text_to_image(text, size_type, openai_api_key)
base64_data = await oas3_openai_text_to_image(text, size_type)
else:
raise ValueError("Missing necessary parameters.")