moved integrations to separate folder, added a .env which will take c…

…are of everything you need, and now have a reddit app where you can get a random subreddit post or a specified reddit post

_integrations/.env

@@ -0,0 +1,5 @@
+BASE_ID=1111
+API_KEY=1111
+OPENAI_API_KEY=sk-proj-1111
+CLIENT_ID=1111
+CLIENT_SECRET=1111

_integrations/README.md

@@ -0,0 +1,25 @@
+# **Autogen + Integrations**
+One thing I think is somewhat missing from Autogen is API Integration with other...well API's.  This is something Langchain does well.  But my goal is to add more than what they offer for some of these API's.
+
+## Imports Needed:
+### Airtable
+- pyautogen
+- pyairtable
+### Reddit
+- pyautogen
+- praw (this stands for **Python Reddit API Wrapper**)
+
+## .env
+- BASE_ID=airtable base id
+- API_KEY=airtable api key (may re-name this in future)
+- OPENAI_API_KEY=openai's api key
+- CLIENT_ID=the id after creating reddit app
+- CLIENT_SECRET=the secret after creating reddit app
+
+## YouTube Link:
+- coming soon
+
+## Updates:
+- 05/30/2024: Added Airtable Integration
+- 06/02/2024: Added Reddit Integration
+

_integrations/reddit/reddit.py

@@ -0,0 +1,105 @@
+import json
+import os
+
+import autogen
+from dotenv import load_dotenv
+from typing_extensions import Annotated
+
+from _integrations.reddit.reddit_classes import RedditPostsLoader
+
+load_dotenv()
+
+config_list = autogen.config_list_from_dotenv(
+    dotenv_file_path="../.env",
+    model_api_key_map={"gpt-3.5-turbo": os.getenv("OPENAI_API_KEY")}
+)
+
+llm_config = {
+    "temperature": 0,
+    "config_list": config_list,
+    "cache_seed": 45
+}
+
+engineer = autogen.AssistantAgent(
+    name="Engineer",
+    llm_config=llm_config,
+    system_message="""
+    I'm Engineer. I'm expert in python programming. I'm executing code tasks required by Admin.
+    """,
+)
+
+user_proxy = autogen.UserProxyAgent(
+    name="Admin",
+    human_input_mode="ALWAYS",
+    code_execution_config=False,
+)
+
+
+@user_proxy.register_for_execution()
+@engineer.register_for_llm(description="Get a random subreddit")
+def get_random_subreddit(
+        mode: Annotated[str, "The mode to retrieve reddit posts from"],
+        number_of_posts: Annotated[int, "The number of posts to retrieve, default should be 1"]
+):
+    loader = RedditPostsLoader(
+        client_id=os.getenv("CLIENT_ID"),
+        client_secret=os.getenv("CLIENT_SECRET"),
+        user_agent="extractor by u/tyler_programming",
+        categories=["new", "hot"],
+        mode=mode,
+        number_posts=number_of_posts,
+    )
+
+    reddit_readers = loader.load()
+
+    serialized_data = json.dumps([reddit.serialize() for reddit in reddit_readers])
+
+    return serialized_data
+
+
+@user_proxy.register_for_execution()
+@engineer.register_for_llm(description="Get a random subreddit")
+def get_specified_subreddit(
+        search_queries: Annotated[str, "The search queries comma separated"],
+        number_of_posts: Annotated[int, "The number of posts to retrieve, default should be 1"]
+):
+    loader = RedditPostsLoader(
+        client_id=os.getenv("CLIENT_ID"),
+        client_secret=os.getenv("CLIENT_SECRET"),
+        user_agent="extractor by u/tyler_programming",
+        categories=["new"],
+        mode="subreddit",
+        search_queries=[
+            search_queries
+        ],
+        number_posts=number_of_posts,
+    )
+
+    reddit_readers = loader.load()
+
+    serialized_data = json.dumps([reddit.serialize() for reddit in reddit_readers])
+
+    return serialized_data
+
+
+chat_result = user_proxy.initiate_chat(
+    engineer,
+    message="""
+        I need you to integrate with Reddit.  Whenever you retrieve a post, you will get a RedditReader Object list and
+        I would like you to format the content and meta_data contained within.
+        
+        Metadata is in this format for each item in the list:
+        
+        reddit_list.post_subreddit,
+        reddit_list.post_category,
+        reddit_list.post_title,
+        reddit_list.post_score,
+        reddit_list.post_id,
+        reddit_list.post_url,
+        reddit_list.post_author
+        
+        Then also add the content from the item.
+        
+        Do not start until I give a command.
+""",
+)

_integrations/reddit/reddit_classes.py

@@ -0,0 +1,139 @@
+from dataclasses import dataclass, asdict, field
+from typing import Optional, Sequence, List, Dict
+
+import praw
+
+
+@dataclass
+class RedditMetaData:
+    post_subreddit: Optional[str] = None
+    post_category: Optional[str] = None
+    post_title: Optional[str] = None
+    post_score: Optional[int] = None
+    post_id: Optional[str] = None
+    post_url: Optional[str] = None
+    post_author: Optional[str] = None
+
+    def serialize(self) -> Dict:
+        return asdict(self)
+
+
+@dataclass
+class RedditReader:
+    content: str = ""
+    meta_data: RedditMetaData = field(default_factory=RedditMetaData)
+
+    def serialize(self) -> Dict:
+        return {
+            'content': self.content,
+            'meta_data': self.meta_data.serialize()
+        }
+
+
+class RedditPostsLoader:
+    """Load `Reddit` posts."""
+
+    def __init__(
+            self,
+            client_id: str,
+            client_secret: str,
+            user_agent: str,
+            search_queries: Sequence[str] = None,
+            mode: Optional[str] = "random",
+            categories: Sequence[str] = ["new"],
+            number_posts: Optional[int] = 10,
+    ):
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.user_agent = user_agent
+        self.search_queries = search_queries
+        self.mode = mode
+        self.categories = categories
+        self.number_posts = number_posts
+
+    def load(self) -> List[RedditReader]:
+        reddit = praw.Reddit(
+            client_id=self.client_id,
+            client_secret=self.client_secret,
+            user_agent=self.user_agent,
+        )
+
+        results: List[RedditReader] = []
+
+        if self.mode == "subreddit":
+            for search_query in self.search_queries:
+                for category in self.categories:
+                    docs = self._subreddit_posts_loader(
+                        search_query=search_query, category=category, reddit=reddit
+                    )
+                    results.extend(docs)
+        if self.mode == "random":
+            for category in self.categories:
+                docs = self._random_posts_loader(
+                    category=category, reddit=reddit
+                )
+                results.extend(docs)
+
+        return results
+
+    def _subreddit_posts_loader(
+            self,
+            search_query: str,
+            category: str,
+            reddit: praw.reddit.Reddit) -> List[RedditReader]:
+        subreddit = reddit.subreddit(search_query)
+        method = getattr(subreddit, category)
+        cat_posts = method(limit=self.number_posts)
+
+        readers = []
+
+        for post in cat_posts:
+            metadata = RedditMetaData(
+                post_subreddit=post.subreddit_name_prefixed,
+                post_category=category,
+                post_title=post.title,
+                post_score=post.score,
+                post_id=post.id,
+                post_url=post.url,
+                post_author=str(post.author)
+            )
+
+            reader = RedditReader(
+                content=post.selftext,
+                meta_data=metadata
+            )
+
+            readers.append(reader)
+
+        return readers
+
+    def _random_posts_loader(
+            self,
+            category: str,
+            reddit: praw.reddit.Reddit) -> list[RedditReader]:
+        subreddit = reddit.random_subreddit()
+        method = getattr(subreddit, category)
+        cat_posts = method(limit=self.number_posts)
+
+        readers = []
+
+        """Format reddit posts into a string."""
+        for post in cat_posts:
+            metadata = RedditMetaData(
+                post_subreddit=post.subreddit_name_prefixed,
+                post_category=category,
+                post_title=post.title,
+                post_score=post.score,
+                post_id=post.id,
+                post_url=post.url,
+                post_author=str(post.author)
+            )
+
+            reader = RedditReader(
+                content=post.selftext,
+                meta_data=metadata
+            )
+
+            readers.append(reader)
+
+        return readers