feat(core): add enumeration and schema modules with utility functions

This commit is contained in:
jinli.yl 2025-12-30 18:16:33 +08:00
parent cb907ea201
commit 266b19ecd0
18 changed files with 608 additions and 4 deletions

View file

@ -1,9 +1,10 @@
from loguru import logger
用英文注释,完善module/class/function docstring,要一句话简洁,不要变更代码
用英文注释,完善module/class/function docstring,要一句话简洁,代码要简洁,符合pep和pylint规范
C0114: Missing module docstring (missing-module-docstring)
C0115: Missing class docstring (missing-class-docstring)
C0116: Missing function or method docstring (missing-function-docstring)
done: { for f in ./*.py; do [[ "$f" != "./__init__.py" ]] && grep -v '^[[:space:]]*#' "$f"; done; } | pbcopy
然后是一个完整的tests,但是不要用其他的包,只是test开头的函数或者类,要求from loguru import logger
然后是一个完整的tests,但是不要用其他的包,只是test开头的函数或者类,要求from loguru import logger

View file

@ -0,0 +1,13 @@
"""enumeration"""
from .chunk_enum import ChunkEnum
from .http_enum import HttpEnum
from .registry_enum import RegistryEnum
from .role import Role
__all__ = [
"ChunkEnum",
"HttpEnum",
"RegistryEnum",
"Role",
]

View file

@ -0,0 +1,25 @@
"""Defines the types of data chunks used in streaming responses."""
from enum import Enum
class ChunkEnum(str, Enum):
"""Enumeration of possible chunk categories for stream processing."""
# Internal reasoning or chain-of-thought process
THINK = "think"
# The final generated response content
ANSWER = "answer"
# Metadata or calls related to external tools
TOOL = "tool"
# Resource consumption and token usage statistics
USAGE = "usage"
# Error messages or exception details
ERROR = "error"
# Final signal indicating the completion of the stream
DONE = "done"

View file

@ -0,0 +1,22 @@
"""Provides a collection of standard HTTP request methods."""
from enum import Enum
class HttpEnum(str, Enum):
"""Enumeration of supported HTTP methods for network requests."""
# Retrieves data from a specified resource
GET = "get"
# Submits data to be processed to a specified resource
POST = "post"
# Identical to GET but only retrieves the response headers
HEAD = "head"
# Uploads or replaces the representation of a target resource
PUT = "put"
# Deletes the specified resource from the server
DELETE = "delete"

View file

@ -0,0 +1,28 @@
"""Defines the registry categories for core components of the system."""
from enum import Enum
class RegistryEnum(str, Enum):
"""Enumeration of component types registered within the application lifecycle."""
# Large Language Model interfaces
LLM = "llm"
# Models used for generating vector embeddings
EMBEDDING_MODEL = "embedding_model"
# Databases or storage systems for vector search
VECTOR_STORE = "vector_store"
# Atomic operations or functional units
OP = "op"
# Orchestrated sequences of operations or workflows
FLOW = "flow"
# External APIs or shared internal services
SERVICE = "service"
# Utilities for tracking and limiting token consumption
TOKEN_COUNTER = "token_counter"

View file

@ -0,0 +1,19 @@
"""Defines the participant roles in a chat completion sequence."""
from enum import Enum
class Role(str, Enum):
"""Enumeration of standard personas involved in a conversation flow."""
# High-level instructions to guide the model's behavior
SYSTEM = "system"
# Input or queries provided by the human user
USER = "user"
# Responses or messages generated by the AI model
ASSISTANT = "assistant"
# Output or results returned from external tool executions
TOOL = "tool"

View file

@ -0,0 +1,40 @@
"""schema"""
from .message import ContentBlock, Message, Trajectory
from .request import Request
from .response import Response
from .service_config import (
CmdConfig,
EmbeddingModelConfig,
FlowConfig,
HttpConfig,
LLMConfig,
MCPConfig,
ServiceConfig,
TokenCounterConfig,
VectorStoreConfig,
)
from .stream_chunk import StreamChunk
from .tool_call import ToolAttr, ToolCall
from .vector_node import VectorNode
__all__ = [
"ContentBlock",
"EmbeddingModelConfig",
"FlowConfig",
"HttpConfig",
"LLMConfig",
"MCPConfig",
"Message",
"Request",
"Response",
"ServiceConfig",
"StreamChunk",
"TokenCounterConfig",
"Trajectory",
"ToolAttr",
"ToolCall",
"VectorNode",
"VectorStoreConfig",
"CmdConfig",
]

View file

@ -0,0 +1,112 @@
"""Data models for multi-modal conversation history and LLM interaction trajectories."""
import datetime
import json
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, model_validator
from .tool_call import ToolCall
from ..enumeration import Role
class ContentBlock(BaseModel):
"""Individual unit of multi-modal content like text, images, or video."""
model_config = ConfigDict(extra="allow")
type: str = Field(default="")
content: str | dict | list = Field(default="")
@model_validator(mode="before")
@classmethod
def init_block(cls, data: dict[str, Any]) -> dict[str, Any]:
"""Dynamically maps the type-specific key to the content field."""
content_type = data.get("type", "")
if content_type and content_type in data:
data["content"] = data[content_type]
return data
def simple_dump(self) -> dict[str, Any]:
"""Serializes the block into an API-compatible dictionary format."""
return {
"type": self.type,
self.type: self.content,
**self.model_extra,
}
class Message(BaseModel):
"""Data model for a single dialogue entry including roles and tool interactions."""
name: str | None = Field(default=None)
role: Role = Field(default=Role.USER)
content: str | list[ContentBlock] = Field(default="")
reasoning_content: str = Field(default="")
tool_calls: list[ToolCall] = Field(default_factory=list)
tool_call_id: str = Field(default="")
time_created: str = Field(
default_factory=lambda: datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
)
metadata: dict[str, Any] = Field(default_factory=dict)
def dump_content(self) -> str | list[dict[str, Any]]:
"""Returns content as a raw string or a list of serialized blocks."""
if isinstance(self.content, str):
return self.content
return [block.simple_dump() for block in self.content]
def simple_dump(self, add_reasoning: bool = True) -> dict[str, Any]:
"""Transforms the message into a simplified dictionary for standard APIs."""
result = {"role": self.role.value, "content": self.dump_content()}
if add_reasoning and self.reasoning_content:
result["reasoning_content"] = self.reasoning_content
if self.tool_calls:
result["tool_calls"] = [tc.simple_output_dump() for tc in self.tool_calls]
if self.tool_call_id:
result["tool_call_id"] = self.tool_call_id
return result
def format_message(
self,
index: int | None = None,
add_time: bool = False,
use_name: bool = False,
add_reasoning: bool = True,
add_tools: bool = True,
) -> str:
"""Generates a human-readable string representation of the message."""
prefix = f"round{index} " if index is not None else ""
time_str = f"[{self.time_created}] " if add_time else ""
header = f"{self.name or self.role.value if use_name else self.role.value}:\n"
lines = [f"{prefix}{time_str}{header}"]
if add_reasoning and self.reasoning_content:
lines.append(f"{self.reasoning_content}\n")
if isinstance(self.content, str):
lines.append(self.content)
elif isinstance(self.content, list):
for block in self.content:
text = (
block.content if isinstance(block.content, str) else json.dumps(block.content, ensure_ascii=False)
)
lines.append(str(text))
if add_tools and self.tool_calls:
for tc in self.tool_calls:
lines.append(f" - tool_call={tc.name} params={tc.arguments}")
return "\n".join(lines).strip()
class Trajectory(BaseModel):
"""Sequence of messages representing a full conversation session and its evaluation."""
task_id: str = Field(default="")
messages: list[Message] = Field(default_factory=list)
score: float = Field(default=0.0)
metadata: dict[str, Any] = Field(default_factory=dict)

View file

@ -0,0 +1,17 @@
"""Defines the data structure for processing incoming user requests and message history."""
from typing import List
from pydantic import Field, BaseModel, ConfigDict
from .message import Message
class Request(BaseModel):
"""Represents a structured request payload containing a query, message list, and metadata."""
model_config = ConfigDict(extra="allow")
query: str = Field(default="")
messages: List[Message] = Field(default_factory=list)
metadata: dict = Field(default_factory=dict)

View file

@ -0,0 +1,11 @@
"""Defines the standardized data structure for model output responses."""
from pydantic import Field, BaseModel
class Response(BaseModel):
"""Represents a structured response containing the execution result, status, and metadata."""
answer: str | dict | list = Field(default="")
success: bool = Field(default=True)
metadata: dict = Field(default_factory=dict)

View file

@ -0,0 +1,113 @@
"""Configuration schemas for service components using Pydantic models."""
from typing import Dict, List
from pydantic import BaseModel, Field, ConfigDict
from .tool_call import ToolCall
class MCPConfig(BaseModel):
"""Configuration for Model Context Protocol transport and network settings."""
model_config = ConfigDict(extra="allow")
transport: str = Field(default="")
host: str = Field(default="0.0.0.0")
port: int = Field(default=8001)
class HttpConfig(BaseModel):
"""Configuration for the HTTP server interface and connection lifecycle."""
model_config = ConfigDict(extra="allow")
host: str = Field(default="0.0.0.0")
port: int = Field(default=8001)
timeout_keep_alive: int = Field(default=3600)
limit_concurrency: int = Field(default=1000)
class CmdConfig(BaseModel):
"""Configuration for command-line flow execution parameters."""
model_config = ConfigDict(extra="allow")
flow: str = Field(default="")
class FlowConfig(ToolCall):
"""Configuration for workflow execution, caching, and error handling."""
model_config = ConfigDict(extra="allow")
flow_content: str = Field(default="")
stream: bool = Field(default=False)
raise_exception: bool = Field(default=True)
enable_cache: bool = Field(default=False)
cache_path: str = Field(default="cache/flow")
cache_expire_hours: float = Field(default=0.1)
class LLMConfig(BaseModel):
"""Configuration for Large Language Model backend and model identification."""
model_config = ConfigDict(extra="allow")
backend: str = Field(default="")
model_name: str = Field(default="")
class EmbeddingModelConfig(BaseModel):
"""Configuration for embedding model backends and identity."""
model_config = ConfigDict(extra="allow")
backend: str = Field(default="")
model_name: str = Field(default="")
class VectorStoreConfig(BaseModel):
"""Configuration for vector database storage and associated embeddings."""
model_config = ConfigDict(extra="allow")
backend: str = Field(default="local")
collection_name: str = Field(default="remy")
embedding_model: str = Field(default="default")
class TokenCounterConfig(BaseModel):
"""Configuration for token counting services and model mapping."""
model_config = ConfigDict(extra="allow")
backend: str = Field(default="base")
model_name: str = Field(default="")
class ServiceConfig(BaseModel):
"""Root configuration schema aggregating all service-level settings and components."""
model_config = ConfigDict(extra="allow")
backend: str = Field(default="")
enable_logo: bool = Field(default=True)
language: str = Field(default="")
thread_pool_max_workers: int = Field(default=16)
ray_max_workers: int = Field(default=-1)
disabled_flows: List[str] = Field(default_factory=list)
enabled_flows: List[str] = Field(default_factory=list)
external_mcp: Dict[str, dict] = Field(
default_factory=dict,
description="External MCP Server configuration",
)
mcp: MCPConfig = Field(default_factory=MCPConfig)
http: HttpConfig = Field(default_factory=HttpConfig)
cmd: CmdConfig = Field(default_factory=CmdConfig)
flow: Dict[str, FlowConfig] = Field(default_factory=dict)
llm: Dict[str, LLMConfig] = Field(default_factory=dict)
embedding_model: Dict[str, EmbeddingModelConfig] = Field(default_factory=dict)
vector_store: Dict[str, VectorStoreConfig] = Field(default_factory=dict)
token_counter: Dict[str, TokenCounterConfig] = Field(default_factory=dict)

View file

@ -0,0 +1,14 @@
"""Defines the data structure for individual data packets in a streaming response."""
from pydantic import Field, BaseModel
from ..enumeration import ChunkEnum
class StreamChunk(BaseModel):
"""Represents a single chunk of streamed data including its type, content, and completion status."""
chunk_type: ChunkEnum = Field(default=ChunkEnum.ANSWER)
chunk: str | dict | list = Field(default="")
done: bool = Field(default=False)
metadata: dict = Field(default_factory=dict)

View file

@ -0,0 +1,128 @@
"""Model definitions for MCP tools and LLM tool call interactions."""
import json
from typing import Dict, List, Literal, Optional, Any
from mcp.types import Tool
from pydantic import BaseModel, Field, model_validator, ConfigDict
TOOL_ATTR_TYPE = Literal["string", "array", "integer", "number", "boolean", "object"]
class ToolAttr(BaseModel):
"""Represent attributes for tool parameters in a JSON schema format."""
type: TOOL_ATTR_TYPE = Field(default="string", description="Attribute data type")
description: str = Field(default="", description="Attribute purpose")
required: bool = Field(default=True, description="Whether the attribute is mandatory")
enum: Optional[List[str]] = Field(default=None, description="Allowed values")
items: Dict[str, Any] = Field(default_factory=dict, description="Schema for array items")
model_config = ConfigDict(extra="allow")
def simple_input_dump(self) -> dict:
"""Export attribute as a standard JSON schema property dictionary."""
res: dict = {"type": self.type, "description": self.description}
if self.enum:
res["enum"] = self.enum
if self.items:
res["items"] = self.items
return res
class ToolCall(BaseModel):
"""Handle tool definitions and execution arguments for LLM integrations."""
index: int = Field(default=0)
id: str = Field(default="")
type: str = Field(default="function")
name: str = Field(default="")
arguments: str = Field(default="{}", description="JSON string of execution arguments")
description: str = Field(default="")
input_schema: Dict[str, ToolAttr] = Field(default_factory=dict)
output_schema: Dict[str, ToolAttr] = Field(default_factory=dict)
@model_validator(mode="before")
@classmethod
def init_tool_call(cls, data: Dict[str, Any]) -> Dict[str, Any]:
"""Map raw API response data to the internal ToolCall structure."""
res = data.copy()
t_type = res.get("type", "function")
inner = res.get(t_type, {})
# Extract basic function metadata
for key in ("name", "arguments", "description"):
if key in inner:
res[key] = inner[key]
# Parse JSON schema parameters into ToolAttr objects
params = inner.get("parameters", {})
if params:
props = params.get("properties", {})
reqs = params.get("required", [])
res["input_schema"] = {k: ToolAttr(**v, required=k in reqs) for k, v in props.items()}
return res
@property
def argument_dict(self) -> dict:
"""Parse the arguments string into a dictionary."""
return json.loads(self.arguments)
def check_argument(self) -> bool:
"""Verify if the arguments string is valid JSON."""
try:
_ = self.argument_dict
return True
except (json.JSONDecodeError, TypeError):
return False
@staticmethod
def _build_schema_dict(schema: Dict[str, ToolAttr]) -> dict:
"""Construct a JSON schema object from a dictionary of ToolAttrs."""
return {
"type": "object",
"properties": {k: v.simple_input_dump() for k, v in schema.items()},
"required": [k for k, v in schema.items() if v.required],
}
def simple_input_dump(self) -> dict:
"""Format the tool definition for LLM provider API requests."""
return {
"type": self.type,
self.type: {
"name": self.name,
"description": self.description,
"parameters": self._build_schema_dict(self.input_schema),
},
}
def simple_output_dump(self) -> dict:
"""Format the tool call result for LLM provider API responses."""
return {
"index": self.index,
"id": self.id,
"type": self.type,
self.type: {"arguments": self.arguments, "name": self.name},
}
@classmethod
def from_mcp_tool(cls, tool: Tool) -> "ToolCall":
"""Create a ToolCall instance from an MCP Tool object."""
props = tool.inputSchema.get("properties", {})
reqs = tool.inputSchema.get("required", [])
return cls(
name=tool.name,
description=tool.description or "",
input_schema={k: ToolAttr(**v, required=k in reqs) for k, v in props.items()},
)
def to_mcp_tool(self) -> Tool:
"""Convert the current instance into an MCP Tool object."""
kwargs = {
"name": self.name,
"description": self.description,
"inputSchema": self._build_schema_dict(self.input_schema),
}
if self.output_schema:
kwargs["outputSchema"] = self._build_schema_dict(self.output_schema)
return Tool(**kwargs)

View file

@ -0,0 +1,15 @@
"""Defines the data structure for individual vector embedding nodes within a retrieval system."""
from typing import List, Dict
from uuid import uuid4
from pydantic import BaseModel, Field
class VectorNode(BaseModel):
"""Represents a discrete unit of text content paired with its corresponding vector embedding and metadata."""
vector_id: str = Field(default_factory=lambda: uuid4().hex)
content: str = Field(default="")
vector: List[float] | None = Field(default=None)
metadata: Dict[str, str | bool | int | float] = Field(default_factory=dict)

View file

@ -1,5 +1,6 @@
"""utils"""
from .timer import timer
from .singleton import singleton
__all__ = ["timer"]
__all__ = ["timer", "singleton"]

View file

@ -0,0 +1,28 @@
"""Case conversion utility for PascalCase, camelCase, and snake_case."""
import re
# Acronyms that should remain uppercase in Pascal/camelCase
_ACRONYMS = {"LLM", "API", "URL", "HTTP", "JSON", "XML", "AI", "MCP"}
_ACRONYM_MAP = {word.lower(): word for word in _ACRONYMS}
def camel_to_snake(content: str) -> str:
"""Convert PascalCase or camelCase to snake_case."""
# Normalize acronyms to title case (e.g., LLM -> Llm) to assist regex splitting
for word in _ACRONYMS:
content = content.replace(word, word.capitalize())
# Insert underscores between case transitions and convert to lowercase
return re.sub(r"(?<!^)(?=[A-Z])", "_", content).lower()
def snake_to_camel(content: str) -> str:
"""Convert snake_case to PascalCase (preserving defined acronyms)."""
return "".join(_ACRONYM_MAP.get(part.lower(), part.capitalize()) for part in content.split("_") if part)
if __name__ == "__main__":
# Quick verification
print(camel_to_snake("OpenAILLMClient")) # open_ai_llm_client
print(snake_to_camel("open_ai_llm_client")) # OpenAILLMClient

View file

@ -0,0 +1,17 @@
"""Module providing a decorator to implement the Singleton design pattern."""
def singleton(cls):
"""A class decorator that ensures only one instance of a class exists."""
# Dictionary to cache the single instance of the class
_instance = {}
def _singleton(*args, **kwargs):
"""Return the existing instance or create a new one if it doesn't exist."""
if cls not in _instance:
# Create and store the instance if it's the first call
_instance[cls] = cls(*args, **kwargs)
return _instance[cls]
return _singleton

View file

@ -45,7 +45,7 @@ def timer(func: F) -> F:
# Use patch to inject metadata instead of relying on stack depth
logger.patch(patcher).info(
"========== cost={:.6f}s ==========",
duration
duration,
)
@functools.wraps(func)
@ -58,7 +58,7 @@ def timer(func: F) -> F:
duration = time.perf_counter() - start_time
logger.patch(patcher).info(
"========== cost={:.6f}s ==========",
duration
duration,
)
if inspect.iscoroutinefunction(func):