Merge pull request longevity-genie#2 from winternewt/manager

Manager

examples/agent_from_yaml.py

@@ -0,0 +1,26 @@
+from dotenv import load_dotenv
+from pathlib import Path
+
+from examples.tools.weather import get_current_weather
+
+import just_agents.llm_options
+from just_agents.just_agent import JustAgent
+from just_agents.just_profile import JustAgentProfile
+
+load_dotenv(override=True)
+
+config_path=Path("./examples/yaml_initialization_example_new.yaml")
+
+created_agent = JustAgent(
+    llm_options=just_agents.llm_options.OPENAI_GPT4oMINI,
+    config_path=config_path,
+    tools=[get_current_weather]
+)
+
+created_agent.save_to_yaml("SimpleWeatherAgent")
+
+loaded_agent = JustAgent.from_yaml("SimpleWeatherAgent", file_path=config_path)
+result = loaded_agent.query(
+    "What's the weather like in San Francisco, Tokyo, and Paris?"
+)
+print (result)

examples/tools/weather.py

@@ -0,0 +1,18 @@
+import json
+
+def get_current_weather(location: str) -> str:
+    """Gets the current weather in a given location.
+    Args:
+        location (str): The name of the location to get the weather for.
+    Returns:
+        str: A JSON string containing the location, temperature, and unit of measurement.
+    """
+    print("Function was actually called! with location: ", location, "")
+    if "tokyo" in location.lower():
+        return json.dumps({"location": "Tokyo", "temperature": "10", "unit": "celsius"})
+    elif "san francisco" in location.lower():
+        return json.dumps({"location": "San Francisco", "temperature": "72", "unit": "fahrenheit"})
+    elif "paris" in location.lower():
+        return json.dumps({"location": "Paris", "temperature": "22", "unit": "celsius"})
+    else:
+        return json.dumps({"location": location, "temperature": "unknown"})

examples/yaml_initialization_example_new.yaml

@@ -0,0 +1,22 @@
+description: "This sample file wad auto-generated using 'agent_from_yaml.py'"
+agent_profiles:
+  SimpleWeatherAgent:
+    class_qualname: just_agents.just_agent.JustAgent
+    description: Generic all-purpose AI agent
+    llm_options:
+      model: gpt-4o-mini
+      temperature: 0.0
+    system_prompt: You are a helpful AI assistant
+    tools:
+    - auto_refresh: true
+      description: "Gets the current weather in a given location.\n    Args:\n   \
+        \     location (str): The name of the location to get the weather for.\n \
+        \   Returns:\n        str: A JSON string containing the location, temperature,\
+        \ and unit of measurement.\n    "
+      function: get_current_weather
+      package: examples.tools.weather
+      parameters:
+      - location:
+          default: null
+          kind: POSITIONAL_OR_KEYWORD
+          type_annotation: <class 'str'>

just_agents/interfaces/IAgent.py

@@ -1,5 +1,6 @@
 from pathlib import Path
-from typing import AsyncGenerator, Any
+from typing import AsyncGenerator, Any, Union, Sequence, Dict
+from abc import ABC, abstractmethod
 from just_agents.utils import resolve_agent_schema
 
 def build_agent(agent_schema: str | Path | dict):
@@ -12,9 +13,35 @@ def build_agent(agent_schema: str | Path | dict):
     elif class_name == "ChainOfThoughtAgent":
         return ChainOfThoughtAgent(agent_schema=agent_schema)
 
-class IAgent:
-    def stream(self, input: str | dict | list[dict]) -> AsyncGenerator[Any, None]:
-        raise NotImplementedError("You need to impelement stream() first!")
+class IAgent(ABC):
 
-    def query(self, input: str | dict | list[dict]) -> str:
-        raise NotImplementedError("You need to impelement query() first!")
+    # @staticmethod
+    # def build(agent_schema: dict):
+    #     import importlib
+    #     try:
+    #         package_name = agent_schema.get("package", None)
+    #         class_name = agent_schema.get("class", None)
+    #
+    #         if package_name is None:
+    #             raise ValueError("Error package_name field should not be empty in agent_schema param during IAgent.build() call.")
+    #         if class_name is None:
+    #             raise ValueError("Error class_name field should not be empty in agent_schema param during IAgent.build() call.")
+    #         # Dynamically import the package
+    #         package = importlib.import_module(package_name)
+    #         # Get the class from the package
+    #         class_ = getattr(package, class_name)
+    #         # Create an instance of the class
+    #         instance = class_(agent_schema=agent_schema)
+    #
+    #         return instance
+    #     except (ImportError, AttributeError) as e:
+    #         print(f"Error creating instance of {class_name} from {package_name}: {e}")
+    #         return None
+
+    @abstractmethod
+    def stream(self, query_input: Union[str, Dict, Sequence[Dict]]) -> AsyncGenerator[Any, None]:
+        raise NotImplementedError("You need to implement stream() first!")
+
+    @abstractmethod
+    def query(self, query_input: Union[str, Dict, Sequence[Dict]]) -> str:
+        raise NotImplementedError("You need to implement query() first!")

just_agents/just_agent.py

@@ -0,0 +1,28 @@
+from pydantic import  Field, PrivateAttr
+from typing import Any, Dict, AsyncGenerator, Union, Sequence
+from just_agents.interfaces.IAgent import IAgent
+from just_agents.llm_session import LLMSession
+from just_agents.llm_options import OPENAI_GPT4oMINI
+from just_agents.just_profile import JustAgentProfile
+
+class JustAgent(IAgent, JustAgentProfile):
+    llm_options: Dict[str, Any] = Field(default=OPENAI_GPT4oMINI)
+    _session: LLMSession = PrivateAttr()
+
+    def model_post_init(self, __context: Any) -> None:
+        super().model_post_init(__context)
+        self._session = LLMSession(llm_options=self.llm_options, tools=self.get_tools_callables())
+        self._session.instruct(self.system_prompt)
+
+    def stream(
+        self,
+        prompt: Union[str, Dict | Sequence[Dict]],
+
+    ) -> AsyncGenerator[Any, None]:
+        return self._session.stream(prompt)
+
+    def query(
+        self,
+        prompt: Union[str, Dict | Sequence[Dict]],
+    ) -> str:
+        return self._session.query(prompt)

just_agents/just_profile.py

@@ -0,0 +1,112 @@
+from pathlib import Path
+from pydantic import Field, field_validator, model_validator
+from typing import Optional, List, ClassVar, Tuple, Sequence, Union, Callable, Self
+
+from regex import template
+
+from just_agents.just_serialization import JustSerializable
+from just_agents.just_tool import JustTool
+
+class JustAgentProfile(JustSerializable):
+    """
+    A Pydantic model representing an agent profile with extended attributes.
+    """
+    DEFAULT_GENERIC_PROMPT: ClassVar[str] = "You are a helpful AI assistant"
+    DEFAULT_DESCRIPTION: ClassVar[str] = "Generic all-purpose AI agent"
+    DEFAULT_PARENT_SECTION: ClassVar[str] = 'agent_profiles'
+    DEFAULT_CONFIG_PATH: ClassVar[Path] = Path('config/agent_profiles.yaml')
+    config_parent_section: Optional[Path] = Field(DEFAULT_PARENT_SECTION, exclude=True)
+
+    system_prompt: str = Field(
+        DEFAULT_GENERIC_PROMPT,
+        description="System prompt of the agent")
+    """System prompt of the agent."""
+
+    description: str = Field(
+        DEFAULT_DESCRIPTION,
+        description="Short description of what the agent does")
+    """Short description of what the agent does."""
+
+    TO_REFRESH: ClassVar[Tuple[str, ...]] = ('shortname', 'description')
+    """Fields to force-renew using LLM"""
+
+    role: Optional[str] = Field(
+        None,
+        description="Role of the agent")
+    """Role of the agent."""
+
+    goal: Optional[str] = Field(
+        None,
+        description="Goal of the agent")
+    """Goal of the agent."""
+
+    task: Optional[str] = Field(
+        None,
+        description="Tasks of the agent")
+    """Tasks of the agent."""
+
+    expertise_domain: Optional[str] = Field(
+        None,
+        description="Agent's field of expertise")
+    """Agent's field of expertise."""
+
+    limitations: Optional[str] = Field(
+        None,
+        description="Agent's known limitations")
+    """Agent's known limitations."""
+
+    backstory: Optional[str] = Field(
+        None,
+        description="Backstory of the agent")
+    """Backstory of the agent."""
+
+    llm_model_name: Optional[str] = Field(
+        None,
+        description="The name of the preferred model to use for inference",
+        alias="model_name" #model_name conflicts with pydantic
+    )
+    """The name of the preferred model to use for inference"""
+
+    tools: Optional[Sequence[Union[Callable|JustTool]]] = Field(
+        None,
+        description="A List[Callable] of tools s available to the agent and their descriptions")
+    """A List[Callable] of tools s available to the agent and their descriptions"""
+
+    knowledge_sources: Optional[List[str]] = Field(
+        None,
+        description="A List[str] of of external knowledge sources the agent is capable of accessing, e.g., databases, APIs, etc.")
+    """List of external knowledge sources the agent is capable of accessing, e.g., databases, APIs, etc."""
+
+    @model_validator(mode='after')
+    def validate_model(self) -> Self:
+        """Converts callables to JustTool instances and refreshes them before validation."""
+        if not self.tools:
+            return self
+        if not isinstance(self.tools, Sequence):
+            raise TypeError("The 'tools' field must be a sequence of callables or JustTool instances.")
+        elif not {x for x in self.tools if not isinstance(x, JustTool)}: #all converted
+            return self
+        new_tools = []
+        for item in self.tools:
+            if isinstance(item, JustTool):
+                new_tools.append(item.refresh())
+            elif callable(item):
+                new_tools.append(JustTool.from_callable(item))
+            else:
+                raise TypeError("Items in 'tools' must be callables or JustTool instances.")
+        setattr(self, 'tools', new_tools)
+        return self
+
+    def get_tools_callables(self) -> Optional[Sequence[Callable]]:
+        """Retrieves the list of callables from the tools."""
+        if self.tools is None:
+            return None
+        return [tool.refresh().get_callable() for tool in self.tools]
+
+    def __str__(self) -> str:
+        """
+        Returns the 'description' field when the instance is converted to a string.
+        """
+        return self.description
+
+