llm-context-providers 0.1.1

No project description provided

LLM Context Providers

A Python package to manage and fetch context from various sources like Asana, Quip, OneNote, Outlook Calendar, etc. This package is designed to be easily extensible, allowing new context providers to be added with minimal changes to the core logic.

Features

• Fetch context from multiple sources asynchronously or synchronously.
• Easily extendable with new context providers.
• Configurable through YAML configuration.
• Handles retries and errors gracefully.

Installation

pip install llm-context-providers

Quick Start

Create an app.py file with the following content to quickly get started:

from llm_context_providers import ContextManager

config = {
    'global': {
        'timezone': 'America/Toronto'
    },
    'context_providers': {
        'asana': {
            'enabled': True,
            'project_id': 'your_asana_project_id',
            'personal_access_token': 'your_personal_access_token',
            'fields': {
                'Task': 'name',
                'Task ID': 'gid',
                'Created At': 'created_at',
                'Modified At': 'modified_at',
                'Completed': 'completed',
                'Assignee': 'assignee',
                'Due On': 'due_on',
                'Notes': 'notes'
            }
        }
    }
}

def main_sync(config):
    manager = ContextManager(config)
    manager.fetch_contexts()
    context = manager.get_combined_context()
    print("Sync Fetch Context (All):\n", context)

if __name__ == "__main__":
    main_sync(config)

Run the application:

python app.py

Full Functionality

For easier maintainability and to handle more complex configurations, it's recommended to use a config.yml file. Here is an example configuration:

global:
  timezone: "America/Toronto"

context_providers:
  asana:
    enabled: true
    project_id: "your_asana_project_id"
    fields:
      Task: "name"
      Task ID: "gid"
      Created At: "created_at"
      Modified At: "modified_at"
      Completed: "completed"
      Assignee: "assignee"
      Due On: "due_on"
      Notes: "notes"
  # Add other context providers here

Using the Configuration File

Update your app.py to load configuration from config.yml and demonstrate the full functionality, including asynchronous and specific context fetching:

import yaml
import asyncio
from llm_context_providers import ContextManager

def load_config(config_file='config.yml'):
    with open(config_file, 'r') as file:
        return yaml.safe_load(file)

async def main_async(config):
    manager = ContextManager(config)
    await manager.fetch_contexts_async()
    context = manager.get_combined_context()
    print("Async Fetch Context (All):\n", context)

async def main_async_specific(config):
    manager = ContextManager(config)
    await manager.fetch_contexts_async(providers=['asana'])
    context = manager.get_combined_context(providers=['asana'])
    print("Async Fetch Context (Specific):\n", context)

def main_sync(config):
    manager = ContextManager(config)
    manager.fetch_contexts()
    context = manager.get_combined_context()
    print("Sync Fetch Context (All):\n", context)

def main_sync_specific(config):
    manager = ContextManager(config)
    manager.fetch_contexts(providers=['asana'])
    context = manager.get_combined_context(providers=['asana'])
    print("Sync Fetch Context (Specific):\n", context)

if __name__ == "__main__":
    config = load_config()

    print("Testing Async Fetch (All):")
    asyncio.run(main_async(config))

    print("\nTesting Async Fetch (Specific):")
    asyncio.run(main_async_specific(config))

    print("\nTesting Sync Fetch (All):")
    main_sync(config)

    print("\nTesting Sync Fetch (Specific):")
    main_sync_specific(config)

Run the application:

python app.py

Extending with New Context Providers

To add a new context provider, create a new class that inherits from ContextProvider and implement the necessary methods:

from llm_context_providers import ContextProvider

class NewContextProvider(ContextProvider):
    def __init__(self, **kwargs):
        super().__init__()
        # Initialize with necessary credentials and configurations

    async def fetch_context_async(self, full_fetch: bool = False):
        # Implement async context fetching logic
        pass

    def fetch_context(self, full_fetch: bool = False):
        # Implement sync context fetching logic
        pass

    def get_context(self):
        # Return the context information
        pass

    async def index_context(self):
        # Implement indexing logic
        pass

    async def search_index(self, query: str):
        # Implement search logic
        pass

    def load_from_index(self, search_results):
        # Implement logic to load search results into context
        pass

Add the new context provider to your configuration file and ensure the package's ContextProvider class can recognize the new provider.

Contributing

Contributions are welcome! Please create a pull request or raise an issue to discuss your ideas.

License

This project is licensed under the MIT License.