Add code

Author: rlancemartin

.env.example

@@ -0,0 +1,3 @@
+ARCADE_API_KEY=
+ARCADE_USER_ID=
+ANTHROPIC_API_KEY=

README.md

@@ -0,0 +1,64 @@
+# Reply gAI
+
+## 🚀 Quickstart with LangGraph server
+
+Sign up for an [Arcade API](https://docs.arcade-ai.com/integrations/toolkits/x) to get access to Twitter data.
+
+Install the langgraph CLI:
+```
+pip install -U "langgraph-cli[inmem]"
+```
+
+Install dependencies:
+```
+pip install -e .
+```
+
+Load API keys into the environment for the LangSmith SDK, Anthropic API and Tavily API:
+```
+export ANTHROPIC_API_KEY=<your_anthropic_api_key>
+export ARCADE_API_KEY=<your_arcade_api_key>
+export ARCADE_USER_ID=<your_arcade_user_id>
+```
+
+Launch the agent:
+```
+langgraph dev
+```
+
+If all is well, you should see the following output:
+
+> Ready!
+
+> API: http://127.0.0.1:2024
+
+> Docs: http://127.0.0.1:2024/docs
+
+> LangGraph Studio Web UI: https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024
+
+## How it works
+
+Reply gAI uses LangGraph to create a workflow that mimics a Twitter user's writing style. Here's how the system operates:
+
+1. **Tweet Collection**
+   - Uses the [Arcade API X Toolkit](https://docs.arcade-ai.com/integrations/toolkits/x) to fetch up to 100 recent tweets from a specified Twitter user
+   - Tweets are stored locally with their text content and URLs
+   - The system automatically refreshes tweets if they're older than the configured age limit
+
+2. **Conversation Flow**
+   - The workflow is managed by a state graph with two main nodes:
+     - `get_tweets`: Fetches and stores recent tweets
+     - `chat`: Generates responses using Claude 3.5 Sonnet
+
+3. **Response Generation**
+   - Claude analyzes the collected tweets to understand the user's writing style
+   - Generates contextually appropriate responses that match the personality and tone of the target Twitter user
+   - Uses a temperature of 0.75 to balance creativity with consistency
+
+4. **Architecture**
+   - Built on LangGraph for workflow management
+   - Uses Anthropic's Claude 3.5 Sonnet for response generation
+   - Integrates with Arcade API for Twitter data access
+   - Maintains conversation state and tweet storage for efficient operation
+
+The system automatically determines whether to fetch new tweets or use existing ones based on their age, ensuring responses are generated using recent and relevant data.

langgraph.json

@@ -0,0 +1,11 @@
+{
+    "dockerfile_lines": [],
+    "graphs": {
+      "reply_gai": "./src/agent/graph.py:graph"
+    },
+    "python_version": "3.11",
+    "env": "./.env",
+    "dependencies": [
+      "."
+    ]
+  }

pyproject.toml

@@ -0,0 +1,56 @@
+[project]
+name = "reply-gAI"
+version = "0.0.1"
+description = "Chat persona based on a Twitter user."
+authors = [
+    { name = "Lance Martin" }
+]
+readme = "README.md"
+license = { text = "MIT" }
+requires-python = ">=3.9"
+dependencies = [
+    "langgraph>=0.2.55",
+    "langchain-community>=0.3.9",
+    "langchain-anthropic>=0.3.0",
+    "arcade_x>=0.1.5",
+]
+
+[project.optional-dependencies]
+dev = ["mypy>=1.11.1", "ruff>=0.6.1"]
+
+[build-system]
+requires = ["setuptools>=73.0.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools]
+packages = ["agent"]
+
+[tool.setuptools.package-dir]
+"agent" = "src/agent"
+
+[tool.setuptools.package-data]
+"*" = ["py.typed"]
+
+[tool.ruff]
+lint.select = [
+    "E",    # pycodestyle
+    "F",    # pyflakes
+    "I",    # isort
+    "D",    # pydocstyle
+    "D401", # First line should be in imperative mood
+    "T201",
+    "UP",
+]
+lint.ignore = [
+    "UP006",
+    "UP007",
+    "UP035",
+    "D417",
+    "E501",
+]
+
+[tool.ruff.lint.per-file-ignores]
+"tests/*" = ["D", "UP"]
+
+[tool.ruff.lint.pydocstyle]
+convention = "google"

src/agent/__init__.py

@@ -0,0 +1,29 @@
+import os
+from dataclasses import dataclass, field, fields
+from typing import Any, Optional
+
+from langchain_core.runnables import RunnableConfig
+from typing_extensions import Annotated
+from dataclasses import dataclass
+
+@dataclass(kw_only=True)
+class Configuration:
+    """The configurable fields for the chatbot."""
+    username: str = "elonmusk"
+    update_tweet: bool = False
+    max_tweet_age_seconds: int = 86400  # 24 hours (24 * 60 * 60 seconds)
+
+    @classmethod
+    def from_runnable_config(
+        cls, config: Optional[RunnableConfig] = None
+    ) -> "Configuration":
+        """Create a Configuration instance from a RunnableConfig."""
+        configurable = (
+            config["configurable"] if config and "configurable" in config else {}
+        )
+        values: dict[str, Any] = {
+            f.name: os.environ.get(f.name.upper(), configurable.get(f.name))
+            for f in fields(cls)
+            if f.init
+        }
+        return cls(**{k: v for k, v in values.items() if v})

src/agent/configuration.py

@@ -0,0 +1,166 @@
+import uuid
+import os
+from datetime import datetime, timezone
+from langchain_core.runnables import RunnableConfig
+from langchain_core.messages import SystemMessage
+from langchain_anthropic import ChatAnthropic 
+from langgraph.graph import MessagesState
+from langgraph.store.base import BaseStore
+from langgraph.graph import END, StateGraph
+from arcadepy import Arcade
+
+import agent.configuration as configuration
+from agent.prompts import CHAT_INSTRUCTIONS
+
+def get_tweets(state: MessagesState, config: RunnableConfig, store: BaseStore) -> dict:
+    """Fetch and store recent tweets for a specified Twitter user.
+    
+    This function authenticates with the Arcade API, retrieves recent tweets for a given
+    username, and stores them in the provided BaseStore instance. Each tweet is stored
+    with its text content and URL.
+    
+    Args:
+        state (MessagesState): Current conversation state (unused but required by graph)
+        config (RunnableConfig): Configuration object containing settings like username
+        store (BaseStore): Storage interface for saving retrieved tweets
+        
+    Returns:
+        dict: Empty dictionary (function stores tweets but doesn't return them)
+        
+    Note:
+        - Requires ARCADE_USER_ID environment variable to be set
+        - Fetches up to 100 most recent tweets from the last 7 days
+        - Stores tweets using (username, "tweets") as namespace
+    """
+
+    # Get the configuration
+    configurable = configuration.Configuration.from_runnable_config(config)
+
+    client = Arcade()  
+    USER_ID = os.environ["ARCADE_USER_ID"]
+    TOOL_NAME = "X.SearchRecentTweetsByUsername"
+
+    auth_response = client.tools.authorize(
+        tool_name=TOOL_NAME,
+        user_id=USER_ID,
+    )
+
+    if auth_response.status != "completed":
+        print(f"Click this link to authorize: {auth_response.authorization_url}")
+
+    # Wait for the authorization to complete
+    client.auth.wait_for_completion(auth_response)
+
+    # Search for recent tweets (last 7 days) on X (Twitter)
+    username = configurable.username
+    # TODO: Check with Arcade about max_results
+    inputs = {"username": username, "max_results": 100}
+    response = client.tools.execute(
+        tool_name=TOOL_NAME,
+        inputs=inputs,
+        user_id=USER_ID,
+    )
+
+    # Format tweets into a string
+    tweets = response.output.value['data']
+
+    # Load the tweet  
+    namespace_for_memory = (username, "tweets")
+    for tweet in tweets:
+        memory_id = tweet.get('id',uuid.uuid4())
+        text = tweet.get('text',"Tweet empty")
+        url = tweet.get('tweet_url',"URL not found")
+        store.put(namespace_for_memory, memory_id, {"text": text,"url": url})
+
+def chat(state: MessagesState, config: RunnableConfig, store: BaseStore) -> dict:
+    """Generate a chat response in the style of a specific Twitter user.
+    
+    This function retrieves tweets from the store for a given username, formats them,
+    and uses them as context for Claude to generate a response that mimics the user's
+    writing style and personality.
+
+    Args:
+        state (MessagesState): Current conversation state containing message history
+        config (RunnableConfig): Configuration object containing settings like username
+        store (BaseStore): Storage interface for accessing saved tweets
+
+    Returns:
+        dict: Contains the generated message in the 'messages' key
+    """
+
+    # Get the configuration
+    configurable = configuration.Configuration.from_runnable_config(config)
+    username = configurable.username
+    
+    # Get the tweets
+    namespace_for_memory = (username, "tweets")
+
+    # Get all the tweets
+    memories = []
+    while mems := store.search(namespace_for_memory, limit=200, offset=len(memories)):
+        memories.extend(mems)
+        
+    # Format the tweets
+    formatted_output = ""
+    for memory in memories:
+        tweet = memory.value
+        formatted_output += f"@{username}: {tweet['text']}\n"
+        formatted_output += "-" * 80 + "\n"
+
+    # Generate a response
+    claude_3_5_sonnet = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0.75) 
+    chat_instructions_formatted = CHAT_INSTRUCTIONS.format(username=username,tweets=formatted_output)
+    msg = claude_3_5_sonnet.invoke([SystemMessage(content=chat_instructions_formatted)]+state['messages'])
+    return {"messages": [msg]} 
+
+def route_to_tweet_loader(state: MessagesState, config: RunnableConfig, store: BaseStore) -> dict:
+    """Route the workflow based on tweet availability and age.
+    
+    This function determines whether to fetch new tweets or proceed to chat by checking:
+    1. If tweets exist for the user in the store
+    2. If existing tweets are too old (beyond max_tweet_age_seconds)
+    
+    Args:
+        state (MessagesState): Current conversation state
+        config (RunnableConfig): Configuration containing username and tweet age settings
+        store (BaseStore): Storage interface for accessing saved tweets
+        
+    Returns:
+        str: Either "get_tweets" to fetch new tweets or "chat" to proceed with conversation
+    """
+
+     # Get the configuration
+    configurable = configuration.Configuration.from_runnable_config(config)
+    username = configurable.username
+    
+    # If we have Tweets from the user, go to chat
+    namespace_for_memory = (username, "tweets")
+    memories = store.search(namespace_for_memory, limit=200)
+
+    # If we have tweets, check if they're too old
+    if memories:    
+        # Get most recent tweet timestamp
+        most_recent = max(mem.created_at for mem in memories)
+        
+        # Calculate time difference
+        now = datetime.now(timezone.utc)
+        time_delta = now - most_recent
+        
+        # If tweets are too old, get new ones
+        if time_delta.total_seconds() > configurable.max_tweet_age_seconds:
+            return "get_tweets"
+        return "chat"
+    # If no tweets for the user, get them 
+    else:
+        return "get_tweets"
+
+# Create the graph + all nodes
+builder = StateGraph(MessagesState, config_schema=configuration.Configuration)
+builder.add_node("chat",chat)
+builder.add_node("get_tweets",get_tweets)
+builder.set_conditional_entry_point(route_to_tweet_loader, ["chat", "get_tweets"])
+builder.add_edge("get_tweets", "chat")
+builder.add_edge("chat", END)
+
+# Compile the graph
+graph = builder.compile()

src/agent/graph.py

@@ -0,0 +1,14 @@
+CHAT_INSTRUCTIONS = """Hey! You're chatting as @{username} on Twitter. I've pulled some of their recent tweets to help you get their vibe:
+
+{tweets}
+
+Quick tips for keeping it real:
+- Jump right into answers naturally, no need for formal intros
+- Chat like you're talking to a friend - keep it casual and fun
+- Use their usual phrases and way of talking
+- Share opinions and hot takes in their style
+- If it's a new topic, just flow with it naturally using their voice
+
+Just vibe like @{username} would - keep it real and conversational!
+
+Here's what they're asking:"""