scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-10-07 03:00:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

style(core): format code according to team style guide

Browse source
This commit is contained in:
jinli.yl 2026-01-05 14:41:22 +08:00
parent afa3196988
commit 36e88b26dd
11 changed files with 46 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
reme_ai/core/context/service_context.py
View file

@ -22,10 +22,21 @@ class ServiceContext(BaseContext):
self.thread_pool: ThreadPoolExecutor | None = None
self.vector_store_dict: dict[str, dict] = {}
self.mcp_server_tool_call_mapping: dict = {}
# Initialize a registry for every category defined in RegistryEnum
self.registry_dict: dict[RegistryEnum, Registry] = {v: Registry() for v in RegistryEnum.__members__.values()}
self.flow_dict: dict = {}
def _update_config_section(self, section_name: str, update_dict: dict | None):
if not update_dict:
return
target_registry = getattr(self.service_config, section_name)
if "default" not in target_registry:
raise KeyError(f"Default `{section_name}` config not found in service_config")
current_config = target_registry["default"]
target_registry["default"] = current_config.model_copy(update=update_dict, deep=True)
def register(self, name: str, register_type: RegistryEnum):
"""Return a decorator to register a component within a specific registry category."""
return self.registry_dict[register_type].register(name=name)

1
reme_ai/core/schema/service_config.py
View file

@ -1,4 +1,5 @@
"""Configuration schemas for service components using Pydantic models."""
import os
from typing import Dict, List

1
reme_ai/core/service/__init__.py
View file

@ -1,4 +1,5 @@
"""service"""
from .base_service import BaseService
from .cmd_service import CmdService
from .http_service import HttpService

5
reme_ai/core/service/cmd_service.py
View file

@ -1,7 +1,5 @@
"""Command service module for managing and executing command-based workflows."""
from typing import Any
from loguru import logger
from .base_service import BaseService
@ -22,7 +20,6 @@ class CmdService(BaseService):
def integrate_flow(self, flow: BaseFlow) -> str | None:
"""Integrate the workflow configuration into the command service."""
self._cmd_flow = CmdFlow(flow=C.service_config.flow)
return None
def run(self) -> None:
"""Execute the command flow in either asynchronous or synchronous mode."""
@ -30,7 +27,7 @@ class CmdService(BaseService):
if self._cmd_flow.async_mode:
response = run_coro_safely(
self._cmd_flow.call(**C.service_config.cmd.model_extra)
self._cmd_flow.call(**C.service_config.cmd.model_extra),
)
else:
response = self._cmd_flow.call_sync(**C.service_config.cmd.model_extra)

2
reme_ai/core/service/http_service.py
View file

@ -43,7 +43,7 @@ class HttpService(BaseService):
self.app.post(
path=f"/{tool_call.name}",
response_model=Response,
description=tool_call.description
description=tool_call.description,
)(execute_endpoint)
return tool_call.name

14
reme_ai/core/service/mcp_service.py
View file

@ -32,12 +32,14 @@ class MCPService(BaseService):
response = await flow.call(**request_instance.model_dump(exclude_none=True))
return response.answer
self.mcp.add_tool(FunctionTool(
name=tool_call.name,
description=tool_call.description,
fn=execute_tool,
parameters=tool_call.parameters.simple_input_dump(),
))
self.mcp.add_tool(
FunctionTool(
name=tool_call.name,
description=tool_call.description,
fn=execute_tool,
parameters=tool_call.parameters.simple_input_dump(),
),
)
return tool_call.name
def run(self):

2
reme_ai/core/utils/common_utils.py
View file

@ -17,4 +17,4 @@ def run_coro_safely(coro: Coroutine[Any, Any, Any]) -> Any | asyncio.Task[Any]:
else:
# Schedule the coroutine as a background task in the active loop
return loop.create_task(coro)
return loop.create_task(coro)

2
reme_ai/core/utils/llm_utils.py
View file

@ -37,4 +37,4 @@ def extract_content(text: str, language_tag: str = "json", greedy: bool = False)
except json.JSONDecodeError:
result = None
return result
return result

2
reme_ai/core/utils/logo_utils.py
View file

@ -81,4 +81,4 @@ def print_logo(service_config: "ServiceConfig"):
expand=False,
)
Console().print(Group("\n", panel, "\n"))
Console().print(Group("\n", panel, "\n"))

36
reme_ai/core/utils/pydantic_config_parser.py
View file

@ -17,7 +17,7 @@ class PydanticConfigParser:
def __init__(self, config_class: type[T], default_config: str = "default"):
"""Initialize parser with a Pydantic config class.
Args:
config_class: Pydantic BaseModel class to validate configs against.
default_config: Default config file name to use if not specified in args.
@ -69,13 +69,13 @@ class PydanticConfigParser:
@staticmethod
def load_from_yaml(yaml_path: str | Path) -> dict:
"""Load configuration from YAML file.
Args:
yaml_path: Path to YAML configuration file.
Returns:
Dictionary containing configuration data.
Raises:
FileNotFoundError: If YAML file does not exist.
"""
@ -90,10 +90,10 @@ class PydanticConfigParser:
def merge_configs(self, *config_dicts: dict) -> dict:
"""Merge multiple config dictionaries in order.
Args:
*config_dicts: Variable number of config dictionaries to merge.
Returns:
Merged configuration dictionary.
"""
@ -104,10 +104,10 @@ class PydanticConfigParser:
def parse_dot_notation(self, dot_list: list[str]) -> dict:
"""Parse dot notation strings into nested dictionary.
Args:
dot_list: List of strings in format "key.subkey=value".
Returns:
Nested dictionary representation of dot notation.
"""
@ -118,7 +118,7 @@ class PydanticConfigParser:
key_path, value_str = item.split("=", 1)
keys = key_path.split(".")
# Build nested dictionary
current = config_dict
for key in keys[:-1]:
@ -131,13 +131,13 @@ class PydanticConfigParser:
"""Find config file path, trying parser directory first then current directory."""
if not config_name.endswith(".yaml"):
config_name += ".yaml"
# Try parser class directory first
config_path = Path(inspect.getfile(self.__class__)).parent / config_name
if config_path.exists():
logger.info(f"load config={config_path}")
return config_path
# Try current directory
logger.warning(f"config={config_path} not found, try {config_name}")
config_path = Path(config_name)
@ -147,13 +147,13 @@ class PydanticConfigParser:
def parse_args(self, *args: str) -> T:
"""Parse CLI arguments and load configs from YAML files.
Args:
*args: CLI arguments in format "key=value" or "config=file.yaml".
Returns:
Validated Pydantic config instance.
Raises:
ValueError: If no config file is specified.
FileNotFoundError: If specified config file does not exist.
@ -174,8 +174,6 @@ class PydanticConfigParser:
# Use default config if not specified
config = config or self.default_config
if not config:
raise ValueError("add `config=<config_file>` in cmd!")
# Load each config file
for single_config in (c.strip() for c in config.split(",") if c.strip()):
@ -192,17 +190,17 @@ class PydanticConfigParser:
def update_config(self, **kwargs) -> T:
"""Update current config with new values using kwargs.
Args:
**kwargs: Key-value pairs where __ in keys represents nested levels.
Returns:
Updated and validated Pydantic config instance.
"""
# Convert kwargs to dot notation and parse
dot_list = [f"{key.replace('__', '.')}={value}" for key, value in kwargs.items()]
override_config = self.parse_dot_notation(dot_list)
# Merge with existing config
final_config = self.merge_configs(self.config_dict, override_config)
return self.config_class.model_validate(final_config)

2
tests/test_logo.py
View file

@ -1,3 +1,5 @@
"""test logo"""
from reme_ai.core.schema import ServiceConfig, MCPConfig
if __name__ == "__main__":