From b4b1ddb264258bfffccb52967b9edb48895175f4 Mon Sep 17 00:00:00 2001 From: Lance Martin Date: Fri, 6 Dec 2024 16:03:20 -0800 Subject: [PATCH] Add code --- .env.example | 3 + README.md | 64 ++++++++++++++ langgraph.json | 11 +++ pyproject.toml | 56 +++++++++++++ src/agent/__init__.py | 0 src/agent/configuration.py | 29 +++++++ src/agent/graph.py | 166 +++++++++++++++++++++++++++++++++++++ src/agent/prompts.py | 14 ++++ 8 files changed, 343 insertions(+) create mode 100644 .env.example create mode 100644 README.md create mode 100644 langgraph.json create mode 100644 pyproject.toml create mode 100644 src/agent/__init__.py create mode 100644 src/agent/configuration.py create mode 100644 src/agent/graph.py create mode 100644 src/agent/prompts.py diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..a6abc9d --- /dev/null +++ b/.env.example @@ -0,0 +1,3 @@ +ARCADE_API_KEY= +ARCADE_USER_ID= +ANTHROPIC_API_KEY= \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..0c37fbd --- /dev/null +++ b/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= +export ARCADE_API_KEY= +export 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. diff --git a/langgraph.json b/langgraph.json new file mode 100644 index 0000000..2419916 --- /dev/null +++ b/langgraph.json @@ -0,0 +1,11 @@ +{ + "dockerfile_lines": [], + "graphs": { + "reply_gai": "./src/agent/graph.py:graph" + }, + "python_version": "3.11", + "env": "./.env", + "dependencies": [ + "." + ] + } \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b3b930f --- /dev/null +++ b/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" \ No newline at end of file diff --git a/src/agent/__init__.py b/src/agent/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/agent/configuration.py b/src/agent/configuration.py new file mode 100644 index 0000000..bacc0be --- /dev/null +++ b/src/agent/configuration.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}) \ No newline at end of file diff --git a/src/agent/graph.py b/src/agent/graph.py new file mode 100644 index 0000000..e24bea4 --- /dev/null +++ b/src/agent/graph.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() \ No newline at end of file diff --git a/src/agent/prompts.py b/src/agent/prompts.py new file mode 100644 index 0000000..2cf2cf3 --- /dev/null +++ b/src/agent/prompts.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:""" \ No newline at end of file