34377614472
Rami Rustom
5/11/23
Created on May 11, 2023 at 9:42 am
·
threads-blog

Building an AutoGPT assistant in Threads using LangChain and Replit

34472321355.png

This past week at Threads was our first AI hackathon. We prototyped an AutoGPT AI assistant that can use Threads to collaborate with humans.

Our goal was to create infrastructure that allows us to experiment with AI agents quickly within Threads. We used LangChain and Replit to create this rapid prototyping environment.

In this tutorial, I'm gonna walk you through how we built it. If you're in a hurry, find the final code here.

Some background on Agents

An agent is an LLM-backed program that has access to tools, and that can plan to use them in order to achieve a goal. You can read more about them here.

LangChain is an open-source library that makes it really easy to chain together calls to LLMs in order to create complex programs, including agents.

Agents become really powerful when you give them long term memory. Early experiments have shown astonishing results, and open-source implementations like AutoGPT and BabyAGI have gone viral.

First, the demo

In this demo, we have an agent running in Replit that we give the following prompt:

I am working on a project with Rami and Chris. Please find out the current status of our research project, and write a thread status report in channel_id=34470847918

The agent will message Rami and Chris to get a status update from each, then publish a thread that synthesizes the updates it gathered.

Progress: NaN%

Before we get started

The first step is to get a Threads API Key:

  1. If you haven't already, sign up and create an org
  2. Go to https://threads.com/integrations, and under Developer API, click "Generate API Key"

The easiest way to follow along with this demo is to use Replit:

  1. Fork the demo, or create a new repl
  2. Create a new secret called THREADS_API_KEY, and paste the Threads API Key you just created

Getting started: The ThreadsToolkit

Our goal is to create an agent that knows how to use Threads. We have an early public API, and the first step is to teach the agent how to use this API.

LangChain has a great abstraction for this: Toolkits. A Toolkit is a collection of Tools, where each tool is a function that the LLM can invoke, along with a description of when to use this tool.

For example, in order to search the web, you can give the agent a Search Tool that can call a search API with the following description:

Useful for when you need to answer questions about current events. Input should be a search query.

In our case, each Threads API endpoint is a different tool that the LLM can use, and inspired by the Jira Toolkit, I wrote a ThreadsToolkit.

First we start with ThreadsAPI, a wrapper around 3 sample API endpoints: postThread, postChatMessage, and postComment.

from typing import Dict, List
import requests
import os
import json

class ThreadsAPI():
def __init__(self):
self.url = "https://threads.com/api/public"
self.token = os.environ['THREADS_API_KEY']
self.headers = {'Authorization': 'Bearer ' + self.token, 'Content-Type': 'application/json'}
def _post(self, url: str, body: Dict[str, str]={}):
r = requests.post(self.url + url, json=body, headers=self.headers)
if r.status_code == 200:
try:
data = r.json()
return data
except requests.exceptions.RequestsJSONDecodeError:
print(f'Failed to parse response as JSON. Response content was: {r.text}')
else:
err_message = f'Request failed with status code {r.status_code}'
return {'ok': False, 'result': {'error': err_message, 'message': 'Check that you are using the correct id'}}

return None

def create_thread(self, channel_id: str, blocks: List[str]):
return self._post('/postThread', body={'channelID': channel_id, 'blocks': blocks})
def post_chat_message(self, chat_id: str, body: str):
return self._post('/postChatMessage', body={'chatID': chat_id, 'body': body})

def post_comment(self, thread_id: str, parent_id: str = '', blocks: List[str] = []):
return self._post('/postComment', body={'threadID': thread_id, 'parentID': parent_id, 'blocks': blocks})

To explain what each endpoint does to the LLM, we write a brief description that will be injected into the agent prompt:

CREATE_THREAD_PROMPT = """
This tool is a wrapper around Thread's create_thread API, useful when you need to create a new thread.
The input is a dictionary with 2 fields: channel_id (string), and blocks (an array of strings).
The string blocks are formatted using Markdown.
For example, to create a new thread with 3 blocks in channel with channel_id=123, you would pass in the following
dictionary:
{{"channel_id": "123", "blocks": ["# Title", "## A header", "some content"]}}
"""

We then define a generic ThreadsTool class, which encapsulates an API endpoint, and a description like one we wrote above.

from pydantic import Field
from typing import Dict
from langchain.tools.base import BaseTool
from threads.api import ThreadsAPI

class ThreadsTool(BaseTool):
name = ""
description = ""
api: ThreadsAPI = Field(default_factory=ThreadsAPI)
mode: str
def _run(self, instructions) -> str:
"""Use the Threads API tool to run an operation."""
if isinstance(instructions, dict):
instructions = str(instructions)
return self.api.run(self.mode, instructions)
async def _arun(self, query: str) -> str:
"""Use the tool asynchronously."""
raise NotImplementedError("ThreadsAPI does not support async")

Finally we define ThreadsToolkit, which creates a ThreadTool for each endpoint:

from typing import List

from langchain.agents.agent_toolkits.base import BaseToolkit
from langchain.tools import BaseTool
from threads.api import ThreadsAPI
from threads.tool import ThreadsTool

class ThreadsToolkit(BaseToolkit):
"""Threads Toolkit"""
tools: List[BaseTool] = []

@classmethod
def from_threads_api(cls, threads_api: ThreadsAPI) -> "ThreadsToolkit":
actions = threads_api.list()
tools = [
ThreadsTool(
name=action["name"],
description=action["description"],
mode=action["mode"],
api=threads_api,
) for action in actions
]
return cls(tools=tools)

def get_tools(self) -> List[BaseTool]:
"""Get the tools in the toolkit."""
return self.tools

Humans as Tools

Threads is a collaborative environment, and the agent should be able to communicate with others in the org in order to collaborate with them. To do that, inspired by the LangChain HumanInputRun tool, we created a Tool representing each person in the org that the agent can reach out to for help. The tool will message that user, and wait for their response.

HUMANS = [{
'userID': '34470790405',
'name': 'Chat with Rami',
"description":
("This tool is a chat with Rami, a software engineer in your organization."
"The input should be a question for Rami.")
}, {
'userID': '34470789571',
'name':'Chat with Chris',
"description":
("This tool is a chat with Chris, a designer in your organization."
"The input should be a question for Chris.")
}]

Putting it all together

We pass in the tools from the ThreadsToolkit and HumanToolkit, along with a memory vector store, and we're ready to run our agent!

from langchain.vectorstores import FAISS
from langchain.docstore import InMemoryDocstore
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
import faiss

from auto_gpt.agent import AutoGPT
from humans.toolkit import HumanToolkit
from threads.toolkit import ThreadsToolkit
from threads.api import ThreadsAPI

threads_api = ThreadsAPI()
human_toolkit = HumanToolkit.from_threads_api(threads_api)
threads_toolkit = ThreadsToolkit.from_threads_api(threads_api)

embeddings_model = OpenAIEmbeddings()
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings_model.embed_query, index, InMemoryDocstore({}), {})

agent = AutoGPT.from_llm_and_tools(
ai_name="Tailor",
ai_role="Assistant",
tools=human_toolkit.get_tools() + threads_toolkit.get_tools(),
llm=ChatOpenAI(temperature=0, model_name='gpt-4'),
memory=vectorstore.as_retriever()
)
agent.chain.verbose = True

The agent will be able to understand any task that involves members of the org, and be able to use the Threads API to send and receive messages, and post threads.

agent.run(["I am working on a project with Rami and Chris. Please find out the current status of our research project, and write a thread status report in channel_id=34470847918"])

And that's it!

This demo is nowhere near production-ready, but we were able to get to a reasonable prototype in less than an hour, and in a few hundred lines of code. Not bad!

Here's the final Repl:

If you're interested in building AI agents within Threads, then make sure to sign up, and feel free to reach out to myself, Mark, or Abdul on Twitter.

We'd love to hear more from developers on how to make it easy to deploy AI agents to Threads!

©2022 Threads, Inc.
BlogTwitterLog in
Made with 💜 in SF, NYC, TOR, DEN, SEA, AA