Project talkia

.gitignore

@@ -0,0 +1,84 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# Virtual Environments
+venv/
+env/
+ENV/
+env.bak/
+venv.bak/
+.venv/
+
+# IDE/Editor
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS Files
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+# Environment Variables
+.env
+.env.local
+.env.development.local
+.env.test.local
+.env.production.local
+
+# Logs
+*.log
+logs/
+
+# Database
+*.db
+*.sqlite
+*.sqlite3
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pytest
+.pytest_cache/
+.coverage
+htmlcov/
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Temporary files
+*.tmp
+*.temp
+.cache/
+
+# Project specific
+memory.json

agent-memory/main.py

@@ -0,0 +1,67 @@
+from openai import OpenAI
+from dotenv import load_dotenv
+from dotenv import find_dotenv
+import os
+from utils.record_audio import record_audio
+from utils.basemodel2tool import base_model2tool
+from tools.daily_events import DailyEvents
+from datetime import datetime
+import json
+
+load_dotenv(find_dotenv())
+
+client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+
+while True:
+    memory = {
+        "events": [],
+        "interactions": []
+    } if not os.path.exists("memory.json") else json.load(open("memory.json"))
+
+    filename_audio = record_audio()
+
+    audio_file = open(filename_audio, "rb")
+    transcription = client.audio.transcriptions.create(
+        model="whisper-1", 
+        file=audio_file,
+        language="pt"
+    )
+
+    os.remove(filename_audio)
+
+    text = transcription.text
+
+    print(text)
+
+    actual_date = datetime.now().strftime("%d/%m/%Y")
+
+    completion = client.chat.completions.create(
+    model="gpt-4o",
+    messages=[
+        {"role": "developer", "content": f"You are a helpful assistant. You are responsible for remembering events of my life. Today is {actual_date} use this as a reference to remember events. If the event occurred in the past, you should use the date to remember the event using today's date as a reference."},
+        {"role": "assistant", "content": json.dumps(memory)},
+        {"role": "user", "content": text}
+    ],
+    tool_choice="auto",
+    tools=[
+        base_model2tool(DailyEvents)
+        ]
+    )
+
+    if completion.choices[0].message.tool_calls:
+        for tool_call in completion.choices[0].message.tool_calls:
+            if tool_call.function.name == "DailyEvents":
+                daily_events = DailyEvents(**json.loads(tool_call.function.arguments))
+                memory["events"].append(f"Day: {daily_events.date} - {daily_events.events}")
+
+        memory['interactions'].append(f"Human: {text}")
+        memory['interactions'].append(f"Assistant: Evento do dia {daily_events.date} registrado com sucesso, posso te ajudar com mais alguma coisa?")
+        print(f"Evento do dia {daily_events.date} registrado com sucesso, posso te ajudar com mais alguma coisa?")
+
+    if completion.choices[0].message.content:
+        memory['interactions'].append(f"Human: {text}")
+        memory['interactions'].append(f"Assistant: {completion.choices[0].message.content}")
+        print(completion.choices[0].message.content)
+
+    with open("memory.json", "w") as f:
+        json.dump(memory, f)

agent-memory/requirements.txt

@@ -0,0 +1,6 @@
+pyaudio==0.2.14
+wave==0.0.2
+python-dotenv==1.0.1
+openai==1.61.1
+playsound==1.2.2
+pydantic==2.6.1

agent-memory/tools/daily_events.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, Field
+from typing import List
+
+class DailyEvents(BaseModel):
+    """
+    Identifica e registra eventos diários para salvar em uma planilha.
+
+    Args:
+        date (str): Data em que os eventos devem ser identificados, no formato YYYY-MM-DD
+        events (List[Event]): Lista de eventos identificados no dia, cada um contendo título, 
+            descrição e horário
+
+    Returns:
+        str: Confirmação do registro dos eventos
+    """
+
+    date: str = Field(
+        description="Data em que os eventos ocorreram no formato DD/MM/YYYY"
+    )
+
+    events: List[str] = Field(
+        description="Lista de eventos identificados no dia"
+    )

agent-memory/utils/basemodel2tool.py

@@ -0,0 +1,101 @@
+from enum import Enum
+from types import UnionType
+from typing import (
+    List,
+    Literal,
+    Union,
+    get_args,
+    get_origin,
+)
+
+from pydantic import BaseModel
+from pydantic.fields import FieldInfo
+
+
+def get_field_type(field: FieldInfo) -> str:
+    if field.annotation is None:
+        return "any"
+    return get_simple_type_name(field.annotation)
+
+
+def get_simple_type_name(type_hint: type) -> str:
+    """Get simplified OpenAI-compatible type name from Python type hint.
+
+    Args:
+        type_hint (type): Python type hint to convert
+
+    Returns:
+        str: OpenAI-compatible type name
+    """
+    # Add support for | type hint E.g. str | None
+    if get_origin(type_hint) in (Union, UnionType) or isinstance(type_hint, UnionType):
+        types = [t for t in get_args(type_hint) if t is not type(None)]
+        if types:
+            return get_simple_type_name(types[0])
+        raise ValueError(f"Invalid Union type {type_hint}")
+
+    if get_origin(type_hint) in (list, List):
+        return "array"
+    elif type_hint is str:
+        return "string"
+    elif type_hint is int:
+        return "integer"
+    elif type_hint is float:
+        return "number"
+    elif type_hint is bool:
+        return "boolean"
+    elif get_origin(type_hint) == Literal:
+        return "string"  # Literal values will be handled as enum values
+    elif isinstance(type_hint, type) and issubclass(type_hint, BaseModel):
+        return "object"  # Handle nested BaseModel classes
+    else:
+        raise ValueError(f"Type hint {type_hint} not supported.")
+
+
+def base_model2tool(model: type[BaseModel]) -> dict:
+    """Convert a Pydantic BaseModel to OpenAI function format.
+
+    Args:
+        model (type[BaseModel]): Pydantic model class to convert
+
+    Returns:
+        dict: OpenAI function-calling format dictionary
+    """
+    json_output = {
+        "type": "function",
+        "function": {
+            "name": model.__name__,
+            "description": model.__doc__ or "",
+            "parameters": {
+                "type": "object",
+                "properties": {}
+            }
+        }
+    }
+
+    required_properties = []
+
+    for name, field in model.model_fields.items():
+        prop_dict = {
+            "type": get_field_type(field),
+            "description": field.description or "",
+        }
+
+        if get_origin(field.annotation) in (list, List):
+            prop_dict["type"] = "array"
+            args = get_args(field.annotation)
+            if args:
+                prop_dict["items"] = {"type": get_simple_type_name(args[0])}
+
+        if field.is_required():
+            required_properties.append(name)
+
+        if isinstance(field.annotation, type) and issubclass(field.annotation, Enum):
+            prop_dict["enum"] = [str(e.value) for e in field.annotation]
+
+        json_output["function"]["parameters"]["properties"][name] = prop_dict
+
+    if required_properties:
+        json_output["function"]["parameters"]["required"] = required_properties
+
+    return json_output

agent-memory/utils/record_audio.py

@@ -0,0 +1,70 @@
+import pyaudio
+import wave
+import time
+from datetime import datetime
+from typing import Optional
+import sys
+import select
+
+def record_audio() -> Optional[str]:
+    """Records audio from microphone when user presses Enter and stops when Enter is pressed again.
+
+    Returns:
+        str: Path to the saved WAV file, or None if recording failed
+    """
+    CHUNK = 1024
+    FORMAT = pyaudio.paInt16
+    CHANNELS = 1
+    RATE = 44100
+
+    audio = pyaudio.PyAudio()
+    frames = []
+    recording = False
+
+    def callback(in_data, frame_count, time_info, status):
+        if recording:
+            frames.append(in_data)
+        return (in_data, pyaudio.paContinue)
+
+    try:
+        stream = audio.open(
+            format=FORMAT,
+            channels=CHANNELS, 
+            rate=RATE,
+            input=True,
+            frames_per_buffer=CHUNK,
+            stream_callback=callback
+        )
+
+        stream.start_stream()
+        print("Press Enter to start recording, press Enter again to stop...")
+
+        while True:
+            # Check if Enter key was pressed
+            if select.select([sys.stdin], [], [], 0.1)[0]:
+                sys.stdin.readline()
+                if not recording:
+                    recording = True
+                    print("Recording...")
+                else:
+                    break
+            time.sleep(0.1)  # Reduce CPU usage
+
+        stream.stop_stream()
+        stream.close()
+
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        filename = f"recording_{timestamp}.wav"
+
+        with wave.open(filename, 'wb') as wf:
+            wf.setnchannels(CHANNELS)
+            wf.setsampwidth(audio.get_sample_size(FORMAT))
+            wf.setframerate(RATE)
+            wf.writeframes(b''.join(frames))
+
+        return filename
+
+    finally:
+        audio.terminate()
+
+    return None