from typing import Dict, Any, List, Set import json from functools import cached_property from llmops.environment.executor import CompoundExecutor class BlogExpert: """博客系统的业务专家实现""" def __init__(self, swagger_path: str, executor: CompoundExecutor): """初始化BlogExpert Args: swagger_path: OpenAPI/Swagger文件的路径 executor: CompoundExecutor实例 """ self.abs_swagger_path = swagger_path self.executor = executor try: with open(self.abs_swagger_path, "r") as f: self.spec = json.load(f) except ValueError as e: raise ValueError(f"Failed to load OpenAPI/Swagger file: {e}") @property def operation_ids(self) -> List[str]: """获取所有operationId的列表""" operation_ids = [] for path in self.spec.get("paths", {}).values(): for operation in path.values(): if "operationId" in operation: operation_ids.append(operation["operationId"]) return operation_ids @property def extract_ids_and_descriptions(self) -> Dict[str, str]: """获取所有operationId及其描述的映射""" id_descriptions = {} for path in self.spec.get("paths", {}).values(): for operation in path.values(): if "operationId" in operation: id_descriptions[operation["operationId"]] = operation.get( "summary", "" ) return id_descriptions def _collect_refs(self, obj: Any, refs: Set[str]): """递归收集对象中的所有$ref引用 Args: obj: 要检查的对象 refs: 收集到的引用集合 """ if isinstance(obj, dict): if "$ref" in obj: refs.add(obj["$ref"]) for value in obj.values(): self._collect_refs(value, refs) elif isinstance(obj, list): for item in obj: self._collect_refs(item, refs) def _get_schema_name(self, ref: str) -> str: """从$ref字符串中提取schema名称""" return ref.split("/")[-1] def _extract_required_schemas(self, refs: Set[str]) -> Dict[str, Any]: """提取所需的schemas Args: refs: 收集到的$ref引用集合 Returns: 包含所需schemas的字典 """ schemas = {} processed_refs = set() while refs: ref = refs.pop() if ref in processed_refs: continue processed_refs.add(ref) if not ref.startswith("#/components/schemas/"): continue schema_name = self._get_schema_name(ref) schema = self.spec["components"]["schemas"].get(schema_name) if schema: schemas[schema_name] = schema # 收集这个schema中的新引用 new_refs = set() self._collect_refs(schema, new_refs) # 添加新的未处理的引用 refs.update(new_refs - processed_refs) return schemas def extract_operation(self, operation_id: str) -> Dict[str, Any]: """提取指定operationId的操作信息和相关schemas Args: operation_id: 要提取的操作ID Returns: Dict包含两部分: - paths: 包含该操作的路径信息 - components: 包含该操作依赖的所有schemas """ # 查找操作 operation_path = None operation_method = None operation_info = None for path, methods in self.spec.get("paths", {}).items(): for method, operation in methods.items(): if operation.get("operationId") == operation_id: operation_path = path operation_method = method operation_info = operation break if operation_info: break if not operation_info: raise ValueError(f"Operation {operation_id} not found") # 收集所有引用 refs: Set[dict] = set() self._collect_refs(operation_info, refs) # 构建返回结果 result = { "paths": {operation_path: {operation_method: operation_info}}, "components": {"schemas": self._extract_required_schemas(refs)}, } return result # def list_all_resources(self) -> Dict[str, Any]: # """mamually list all resources""" # resources: Dict[str, Any] = { # "posts": {}, # "comments": {}, # "replies": {}, # "tags": {}, # } # list_posts_request = { # "method": "GET", # "path": "/api/posts", # } # posts_response = self.executor.execute_http_request(**list_posts_request) # for post in posts_response.json: # resources["posts"][post["id"]] = {} # resources["posts"][post["id"]]["attributes"] = post # resources["posts"][post["id"]]["state"] = post.get("status") # resources["posts"][post["id"]]["references"] = post.get("tag_ids", []) # resources["posts"][post["id"]]["parent_id"] = None # list_comments_request = { # "method": "GET", # "path": f"/api/posts/{post['id']}/comments", # } # comments_response = self.executor.execute_http_request(**list_comments_request) # if comments_response.status_code == 403: # continue # for comment in comments_response.json: # resources["comments"][comment["id"]] = {} # resources["comments"][comment["id"]]["attributes"] = comment # resources["comments"][comment["id"]]["state"] = comment.get("status") # resources["comments"][comment["id"]]["parent_id"] = post["id"] # resources["comments"][comment["id"]]["references"] = [] # list_replies_request = { # "method": "GET", # "path": f"/api/posts/{post['id']}/comments/{comment['id']}/replies", # } # replies_response = self.executor.execute_http_request(**list_replies_request) # if replies_response.status_code == 403: # continue # for reply in replies_response.json: # resources["replies"][reply["id"]] = {} # resources["replies"][reply["id"]]["attributes"] = reply # resources["replies"][reply["id"]]["state"] = reply.get("status") # resources["replies"][reply["id"]]["parent_id"] = comment["id"] # resources["replies"][reply["id"]]["references"] = [] # list_tags_request = { # "method": "GET", # "path": "/api/tags", # } # tags_response = self.executor.execute_http_request(**list_tags_request) # for tag in tags_response.json: # if tag.get("id"): # resources["tags"][tag["id"]] = {} # resources["tags"][tag["id"]]["attributes"] = tag # resources["tags"][tag["id"]]["state"] = tag.get("status") # resources["tags"][tag["id"]]["parent_id"] = None # resources["tags"][tag["id"]]["references"] = [] # return dict( # posts=resources["posts"], # comments=resources["comments"], # replies=resources["replies"], # tags=resources["tags"], # ) def list_all_resources(self) -> Dict[str, Any]: """mamually initialize a list of resource types""" return { "posts": {}, "comments": {}, "replies": {}, "tags": {}, "reactions": {}, } # def cleanup(self): # """cleanup resources # currently an empty implementation, can be extended later # """ # print("start to clean up resources") # current_state = self.list_all_resources() # print( # "length of current_state: ", # len(current_state["posts"]) + len(current_state["tags"]), # ) # for post_id, _ in current_state["posts"].items(): # self.executor.execute_http_request(method="DELETE", path=f"/api/posts/{post_id}") # for tags in current_state["tags"]: # self.executor.execute_http_request(method="DELETE", path=f"/api/tags/{tags}") # print("resources cleanup completed") @property def biz_system_prompt_preprocessor(self) -> str: """general biz logic for preprocess""" return "" @property def biz_system_prompt_operation_selector(self) -> str: """general biz logic for operation_selector""" return "There are no transitional states (including MODIFYING) in this system. So don't choose an operation in GET method." @property def biz_system_prompt_request_generator(self) -> str: """general biz logic for request_generator""" return "If the field is optional and not required, leave it blank" @property def biz_system_prompt_response_analyzer(self) -> str: """general biz logic for preprocess""" return "" @property def biz_system_prompt_target_matcher(self) -> str: """general biz logic for preprocess""" return ""