add: llm_adapter

src/ai_powered/constants.py

@@ -3,16 +3,27 @@
 DEBUG = os.environ.get('DEBUG', 'False').lower() in {'true', '1', 'yes', 'on'}
 OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "sk-1234567890ab-MOCK-API-KEY")
 OPENAI_BASE_URL = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com")
-OPENAI_MODEL_NAME = os.environ.get("OPENAI_MODEL_NAME", "gpt-4o-mini")
+OPENAI_MODEL_NAME = os.environ.get("OPENAI_MODEL_NAME")
 
 SYSTEM_PROMPT = """
 You are a function simulator,
 your task is to understand the intent of the following function,
 and then begin to simulate its execution,
-that is: the user will repeatedly ask for the results of this function under different actual arguments,
-you need to call the return_result function to return the results to the user.
+that is: the user will ask for the result of this function with actual arguments,
+the arguments is provided in the form of a json string,
+you need to call the return_result function to return the result to the user.
+the result you return should be a valid json string that conforms to the schema of the parameters of the function,
+and without any decoration or additional information.
 
 The function is as follows:
 {signature}
     ''' {docstring} '''
+
+The schema of arguments is as follows:
+{parameters_schema}
+"""
+
+SYSTEM_PROMPT_RETURN_SCHEMA = """
+The schema of expected return value is as follows:
+{return_schema}
 """

src/ai_powered/decorators.py

@@ -1,18 +1,21 @@
 from functools import wraps
-import sys
 from typing import Any, Callable, ParamSpec, TypeVar
-import openai
 import json
 from pydantic import Field, TypeAdapter, create_model
-from .constants import DEBUG, OPENAI_MODEL_NAME, SYSTEM_PROMPT
-from .colors import green
+
+from ai_powered.llm_adapter.definitions import ModelFeature
+from ai_powered.llm_adapter.generic_adapter import GenericFunctionSimulator
+from ai_powered.llm_adapter.known_models import complete_model_config
+from .constants import DEBUG, OPENAI_API_KEY, OPENAI_BASE_URL, OPENAI_MODEL_NAME, SYSTEM_PROMPT, SYSTEM_PROMPT_RETURN_SCHEMA
+from .colors import gray, green
 import inspect
 
 P = ParamSpec("P")
 R = TypeVar("R")
 
 def ai_powered(fn : Callable[P, R]) -> Callable[P, R]:
 
+    function_name = fn.__name__
     sig = inspect.signature(fn)
     docstring = inspect.getdoc(fn)
 
@@ -25,79 +28,62 @@ def ai_powered(fn : Callable[P, R]) -> Callable[P, R]:
 
         print(f"{sig.return_annotation =}")
 
-
     params_ta : dict[str, TypeAdapter[Any]] = {param.name: TypeAdapter(param.annotation) for param in sig.parameters.values()}
+    parameters_schema = {param_name: ta.json_schema() for param_name, ta in params_ta.items()}
     return_ta = TypeAdapter(sig.return_annotation)
 
     if DEBUG:
+        for param_name, schema in parameters_schema.items():
+            print(f"{param_name} (json schema): {schema}")
 
-        for param in sig.parameters.values():
-            print(f"{param.name} (json schema): {params_ta[param.name].json_schema()}")
-
-        print(f"return (json schema): {return_ta.json_schema()}")
+        return_schema = return_ta.json_schema()
+        print(f"return (json schema): {return_schema}")
 
     Result = create_model("Result", result=(sig.return_annotation,Field(...)))
     Result_ta = TypeAdapter(Result)
-
-    sys_prompt = SYSTEM_PROMPT.format(signature = sig, docstring = docstring or "not provided, guess the most possible intention from the function name")
     result_schema = Result_ta.json_schema()
+
+    model_config = complete_model_config(OPENAI_BASE_URL, OPENAI_MODEL_NAME)
+    model_name = model_config.model_name
+    model_features: set[ModelFeature] = model_config.supported_features
+    model_options: dict[str, Any] = model_config.suggested_options
+
+    sys_prompt = SYSTEM_PROMPT.format(
+        signature = sig,
+        docstring = docstring or "not provided, guess the most possible intention from the function name",
+        parameters_schema = parameters_schema,
+    ) + SYSTEM_PROMPT_RETURN_SCHEMA.format(
+        return_schema = result_schema,
+    ) if "function_call" in model_features else ""
+
     if DEBUG:
         print(f"{sys_prompt =}")
         print(f"{result_schema =}")
 
+    fn_simulator = GenericFunctionSimulator(
+        function_name, f"{sig}", docstring, parameters_schema, result_schema,
+        OPENAI_BASE_URL, OPENAI_API_KEY, model_name, model_features, model_options, sys_prompt
+    )
+
     if DEBUG:
-        print(green(f"[fn {fn.__name__}] AI is powering up."))
+        print("fn_simulator =", gray(f"{fn_simulator}"))
+        print(green(f"[fn {function_name}] AI is powering up."))
 
     @wraps(fn)
-    def wrapped_fn(*args: P.args, **kwargs: P.kwargs) -> R:
-        real_argument = sig.bind(*args, **kwargs)
-        real_argument_str = json.dumps(real_argument.arguments)
+    def wrapper_fn(*args: P.args, **kwargs: P.kwargs) -> R:
+        real_arg = sig.bind(*args, **kwargs)
+        real_arg_str = json.dumps(real_arg.arguments)
 
         if DEBUG:
-            print(f"{real_argument_str =}")
-            print(green(f"[fn {fn.__name__}] request prepared."))
-
-        client = openai.OpenAI() # default api_key = os.environ["OPENAI_API_KEY"], base_url = os.environ["OPENAI_BASE_URL"]
-        response = client.chat.completions.create(
-            model = OPENAI_MODEL_NAME,
-            messages = [
-                {"role": "system", "content": sys_prompt},
-                {"role": "user", "content": real_argument_str}
-            ],
-            tools = [{
-                "type": "function",
-                "function": {
-                    "name": "return_result",
-                    "parameters": result_schema,
-                },
-            }],
-            tool_choice = {"type": "function", "function": {"name": "return_result"}},
-        )
+            print(f"{real_arg_str =}")
+            print(green(f"[fn {function_name}] request prepared."))
 
-        if DEBUG:
-            print(f"{response =}")
-            print(green(f"[fn {fn.__name__}] response received."))
+        resp_str = fn_simulator.query_model(real_arg_str)
+        print(green(f"[fn {function_name}] response extracted."))
 
-        resp_msg = response.choices[0].message
-        resp_str = response.choices[0].message.content
-        tool_calls = resp_msg.tool_calls
+        returned_result = Result_ta.validate_json(resp_str)
+        print(green(f"[fn {function_name}] response validated."))
 
-        if DEBUG:
-            print(f"{resp_msg =}")
-            print(f"{resp_str =}")
-            print(f"{tool_calls =}")
-            if tool_calls:
-                for tool_call in tool_calls:
-                    function_name = tool_call.function.name
-                    function_args = json.loads(tool_call.function.arguments)
-                    print(f"{function_name =}")
-                    print(f"{function_args =}")
-
-        assert tool_calls is not None
-        returned_result_str = tool_calls[0].function.arguments
-        print(green(f"[fn {fn.__name__}] response extracted."))
-        returned_result = Result_ta.validate_json(returned_result_str)
-        print(green(f"[fn {fn.__name__}] response validated."))
         return returned_result.result #type: ignore
 
-    return wrapped_fn
+    return wrapper_fn

src/ai_powered/llm_adapter/definitions.py

@@ -0,0 +1,30 @@
+from abc import ABC
+from dataclasses import dataclass
+from typing import Any, Literal, Optional
+
+'''
+    信息确定的过程:
+    首先是连接信息被确定,如BASE URL,API KEY等
+    然后是模型信息被确定,如模型名称，模型参数等
+    模型信息确定后,模型支持的特性随之被确定,如是否支持函数调用,是否支持JSON格式的返回值等
+    再然后是函数信息被确定,如函数签名，函数文档等
+    最终模拟函数执行时,需要参考函数信息和模型信息,以及连接信息,来确定函数执行的具体方式
+'''
+
+#Ref: https://ollama.fan/reference/openai/#supported-features
+ModelFeature = Literal["function_call", "response_json", "specify_seed"]
+ALL_FEATURES : set[ModelFeature] = {"function_call", "response_json", "specify_seed"}
+
+
+@dataclass(frozen=True)
+class FunctionSimulator (ABC):
+    ''' just a wrapper to call model, without checking type correctness '''
+
+    function_name: str
+    signature: str
+    docstring: Optional[str]
+    parameters_schema: dict[str, Any]
+    return_schema: dict[str, Any]
+
+    def execute(self, arguments_json: str) -> str:
+        ...

src/ai_powered/llm_adapter/generic_adapter.py

@@ -0,0 +1,68 @@
+from dataclasses import dataclass
+from typing import Any, Set
+from ai_powered.colors import green
+from ai_powered.constants import DEBUG
+from .definitions import FunctionSimulator, ModelFeature
+import openai
+
+
+@dataclass(frozen=True)
+class GenericFunctionSimulator (FunctionSimulator):
+    ''' implementation of FunctionSimulator for OpenAI compatible models '''
+
+    base_url: str
+    api_key: str
+    model_name: str
+    model_features: Set[ModelFeature]
+    model_options: dict[str, Any]
+    system_prompt : str
+
+    def query_model(self, user_msg: str) -> str:
+        client = openai.OpenAI(base_url=self.base_url, api_key=self.api_key, **self.model_options)
+
+        response = client.chat.completions.create(
+            model = self.model_name,
+            messages = [
+                {"role": "system", "content": self.system_prompt},
+                {"role": "user", "content": user_msg}
+            ],
+            tools = [{
+                "type": "function",
+                "function": {
+                    "name": "return_result",
+                    "parameters": self.return_schema,
+                },
+            }],
+            tool_choice = {"type": "function", "function": {"name": "return_result"}},
+        )
+
+        if DEBUG:
+            print(f"{response =}")
+            print(green(f"[fn {self.function_name}] response received."))
+
+        resp_msg = response.choices[0].message
+        tool_calls = resp_msg.tool_calls
+
+        if tool_calls is not None:
+            return tool_calls[0].function.arguments
+        else:
+            raw_resp_str = resp_msg.content
+            assert raw_resp_str is not None
+
+            # raw_resp_str = "```json\n{"result": 2}\n```"
+
+            if raw_resp_str.startswith("```json\n") and raw_resp_str.endswith("\n```"):
+                unwrapped_resp_str = raw_resp_str[8:-4]
+            else:
+                unwrapped_resp_str = raw_resp_str
+
+            # unwrapped_result_str = "2"
+
+            if unwrapped_resp_str.startswith('{"result":') and unwrapped_resp_str.endswith("}"):
+                result_str = unwrapped_resp_str
+            else:
+                result_str = f'{{"result": {unwrapped_resp_str}}}'
+
+            if DEBUG:
+                print(f"{result_str =}")
+            return result_str

src/ai_powered/llm_adapter/known_models.py

@@ -0,0 +1,71 @@
+from dataclasses import dataclass, field
+from typing import Any, Callable, Optional, Set, TypeAlias
+
+from ai_powered.llm_adapter.definitions import ALL_FEATURES, ModelFeature
+
+@dataclass
+class KnownPlatform:
+    ''' information about a known platform '''
+
+    platform_name: str
+    match_platform_url: Callable[[str], bool]
+    known_model_list: list["KnownModel"]
+
+@dataclass
+class KnownModel:
+    ''' information about a known model '''
+
+    model_name: str
+    supported_features: Set[ModelFeature]
+    suggested_options: dict[str, Any] = field(default_factory=dict)
+
+
+def contains(s: str) -> Callable[[str], bool]:
+    ''' create a function to check if a string contains a substring '''
+    return lambda text: s in text
+
+def starts_with(s: str) -> Callable[[str], bool]:
+    ''' create a function to check if a string starts with a substring '''
+    return lambda text: text.startswith(s)
+
+def equals(s: str) -> Callable[[str], bool]:
+    ''' create a function to check if a string contains a substring '''
+    return lambda text: s == text
+
+KNOWN_PLATFORMS : list[KnownPlatform] = [
+    KnownPlatform(
+        platform_name = "openai",
+        match_platform_url = contains("openai"),
+        known_model_list = [
+            KnownModel("gpt-4o-mini", ALL_FEATURES),
+            KnownModel("gpt-4o", ALL_FEATURES),
+        ]
+    ),
+    KnownPlatform(
+        platform_name = "deepseek",
+        match_platform_url = contains("deepseek"),
+        known_model_list = [
+            KnownModel("deepseek-chat", set()),
+            KnownModel("deepseek-coder", set()),
+        ]
+    ),
+]
+
+ModelConfig : TypeAlias = KnownModel
+
+def complete_model_config(platform_url: str, model_name: Optional[str]) -> ModelConfig:
+    ''' select a known model from a known platform '''
+    for platform in KNOWN_PLATFORMS:
+        if platform.match_platform_url(platform_url):
+            if model_name is not None:
+                for known_model in platform.known_model_list:
+                    if model_name.startswith(known_model.model_name):
+                        return known_model
+            else:
+                return platform.known_model_list[0] #known platform, but model not specified
+            return platform.known_model_list[0] #known platform, but unknown model specified
+    #unknown platform
+    if model_name is not None:
+        return ModelConfig(model_name, ALL_FEATURES)
+    else:
+        raise ValueError(f"Unknown platform: {platform_url}, please specify a model name")