Initial API Setup

This commit is contained in:
Vineeth Voruganti 2023-09-11 00:24:16 -04:00
parent e50900b0a2
commit 320cb47c90
16 changed files with 2456 additions and 0 deletions

15
.env.template Normal file
View File

@ -0,0 +1,15 @@
# Supabase
SUPABASE_URL
SUPABASE_KEY
MEMORY_TABLE
CONVERSATION_TABLE
VECTOR_TABLE
MATCH_FUNCTION
# OpenAI
OPENAI_API_KEY=
OPENAI_API_TYPE=
OPENAI_API_BASE=
OPENAI_API_VERSION=
OPENAI_API_DEPLOYMENT_NAME=

33
cache.py Normal file
View File

@ -0,0 +1,33 @@
"""
Below is an implementation of a basic LRUcache that utilizes the built
in OrderedDict data structure.
"""
from collections import OrderedDict
from mediator import SupabaseMediator
import uuid
from typing import List
from langchain.schema import BaseMessage, Document
from pydantic import BaseModel
class Conversation:
"Wrapper Class for storing contexts between channels. Using an object to pass by reference avoid additional cache hits"
def __init__(self, mediator: SupabaseMediator, user_id: str, conversation_id: str = str(uuid.uuid4()), location_id: str = "web"):
self.mediator: SupabaseMediator = mediator
self.user_id: str = user_id
self.conversation_id: str = conversation_id
self.location_id: str = location_id
def add_message(self, message_type: str, message: BaseMessage,) -> None:
self.mediator.add_message(self.conversation_id, self.user_id, message_type, message)
def messages(self, message_type: str) -> List[BaseMessage]:
return self.mediator.messages(self.conversation_id, self.user_id, message_type)
# vector DB fn
def add_texts(self, texts: List[str]) -> None:
metadatas = [{"conversation_id": self.conversation_id, "user_id": self.user_id} for _ in range(len(texts))]
self.mediator.vector_table.add_texts(texts, metadatas)
# vector DB fn
def similarity_search(self, query: str, match_count: int = 5) -> List[Document]:
return self.mediator.vector_table.similarity_search(query=query, k=match_count, filter={"user_id": self.user_id})

342
chain.py Normal file
View File

@ -0,0 +1,342 @@
import os
from langchain.chat_models import ChatOpenAI, AzureChatOpenAI
from langchain.output_parsers.list import NumberedListOutputParser
from langchain.prompts import load_prompt, ChatPromptTemplate, SystemMessagePromptTemplate
from langchain.schema import AIMessage, HumanMessage, BaseMessage
from dotenv import load_dotenv
from collections.abc import AsyncIterator
from cache import Conversation
from typing import List
load_dotenv()
SYSTEM_THOUGHT = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/thought.yaml'))
SYSTEM_RESPONSE = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/response.yaml'))
SYSTEM_THOUGHT_REVISION = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/thought_revision.yaml'))
SYSTEM_USER_PREDICTION_THOUGHT = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/user_prediction_thought.yaml'))
SYSTEM_USER_PREDICTION_THOUGHT_REVISION = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/user_prediction_thought_revision.yaml'))
SYSTEM_VOE_THOUGHT = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/voe_thought.yaml'))
SYSTEM_VOE = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/voe.yaml'))
SYSTEM_CHECK_VOE_LIST = load_prompt(os.path.join(os.path.dirname(__file__), 'prompts/check_voe_list.yaml'))
class BloomChain:
"Wrapper class for encapsulating the multiple different chains used in reasoning for the tutor's thoughts"
llm: AzureChatOpenAI = AzureChatOpenAI(deployment_name = os.environ['OPENAI_API_DEPLOYMENT_NAME'], temperature=1.2, model_kwargs={"top_p": 0.5})
parser_llm: AzureChatOpenAI = AzureChatOpenAI(deployment_name = os.environ['OPENAI_API_DEPLOYMENT_NAME'])
system_voe_thought: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_VOE_THOUGHT)
system_voe: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_VOE)
system_check_voe_list: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_CHECK_VOE_LIST)
system_thought: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_THOUGHT)
system_thought_revision: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_THOUGHT_REVISION)
system_response: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_RESPONSE)
system_user_prediction_thought: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_USER_PREDICTION_THOUGHT)
system_user_prediction_thought_revision: SystemMessagePromptTemplate = SystemMessagePromptTemplate(prompt=SYSTEM_USER_PREDICTION_THOUGHT_REVISION)
output_parser = NumberedListOutputParser()
def __init__(self) -> None:
pass
@classmethod
def think(cls, cache: Conversation, input: str):
"""Generate Bloom's thought on the user."""
# load message history
thought_prompt = ChatPromptTemplate.from_messages([
cls.system_thought,
*cache.messages("thought"),
HumanMessage(content=input)
])
chain = thought_prompt | cls.llm
cache.add_message("thought", HumanMessage(content=input))
return Streamable(
chain.astream({}, {"tags": ["thought"], "metadata": {"conversation_id": cache.conversation_id, "user_id": cache.user_id}}),
lambda thought: cache.add_message("thought", AIMessage(content=thought))
)
@classmethod
def revise_thought(cls, cache: Conversation, input: str, thought: str):
"""Revise Bloom's thought about the user with retrieved personal data"""
# construct rag prompt, retrieve docs
query = f"input: {input}\n thought: {thought}"
docs = cache.similarity_search(query)
messages = ChatPromptTemplate.from_messages([
cls.system_thought_revision,
*cache.messages('thought_revision'),
HumanMessage(content=input)
])
chain = messages | cls.llm
cache.add_message("thought_revision", HumanMessage(content=input))
return Streamable(
chain.astream({ "thought": thought, "retrieved_vectors": "\n".join(doc.page_content for doc in docs)}, {"tags": ["thought_revision"], "metadata": {"conversation_id": cache.conversation_id, "user_id": cache.user_id}}),
lambda thought_revision: cache.add_message("thought_revision", AIMessage(content=thought_revision)) # add the revised thought to thought memory
)
@classmethod
def respond(cls, cache: Conversation, thought: str, input: str):
"""Generate Bloom's response to the user."""
response_prompt = ChatPromptTemplate.from_messages([
cls.system_response,
*cache.messages("response"),
HumanMessage(content=input)
])
chain = response_prompt | cls.llm
cache.add_message("response", HumanMessage(content=input))
return Streamable(
chain.astream({ "thought": thought }, {"tags": ["response"], "metadata": {"conversation_id": cache.conversation_id, "user_id": cache.user_id}}),
lambda response: cache.add_message("response", AIMessage(content=response))
)
@classmethod
async def think_user_prediction(cls, cache: Conversation, input: str):
"""Generate a thought about what the user is going to say"""
messages = ChatPromptTemplate.from_messages([
cls.system_user_prediction_thought,
])
chain = messages | cls.llm
history = unpack_messages(cache.messages('response'))
user_prediction_thought = await chain.ainvoke(
{"history": history},
{"tags": ["user_prediction_thought"], "metadata": {"conversation_id": cache.conversation_id, "user_id": cache.user_id}}
)
cache.add_message("user_prediction_thought", user_prediction_thought)
return user_prediction_thought.content
@classmethod
async def revise_user_prediction_thought(cls, cache: Conversation, user_prediction_thought: str, input: str):
"""Revise the thought about what the user is going to say based on retrieval of VoE facts"""
messages = ChatPromptTemplate.from_messages([
cls.system_user_prediction_thought_revision,
])
chain = messages | cls.llm
# construct rag prompt, retrieve docs
query = f"input: {input}\n thought: {user_prediction_thought}"
docs = cache.similarity_search(query)
history = unpack_messages(cache.messages('response'))
user_prediction_thought_revision = await chain.ainvoke(
{"history": history, "user_prediction_thought": user_prediction_thought, "retrieved_vectors": "\n".join(doc.page_content for doc in docs)},
config={"tags": ["user_prediction_thought_revision"], "metadata": {"conversation_id": cache.conversation_id, "user_id": cache.user_id}}
)
cache.add_message("user_prediction_thought_revision", user_prediction_thought_revision)
return user_prediction_thought_revision.content
@classmethod
async def think_violation_of_expectation(cls, cache: Conversation, inp: str, user_prediction_thought_revision: str) -> None:
"""Assess whether expectation was violated, derive and store facts"""
# format prompt
messages = ChatPromptTemplate.from_messages([
cls.system_voe_thought
])
chain = messages | cls.llm
voe_thought = await chain.ainvoke(
{"user_prediction_thought_revision": user_prediction_thought_revision, "actual": inp},
config={"tags": ["voe_thought"], "metadata": {"user_id": cache.user_id}}
)
cache.add_message("voe_thought", voe_thought)
return voe_thought.content
@classmethod
async def violation_of_expectation(cls, cache: Conversation, inp: str, user_prediction_thought_revision: str, voe_thought: str) -> None:
"""Assess whether expectation was violated, derive and store facts"""
# format prompt
messages = ChatPromptTemplate.from_messages([
cls.system_voe
])
chain = messages | cls.llm
voe = await chain.ainvoke(
{
"ai_message": cache.messages('response')[-1].content,
"user_prediction_thought_revision": user_prediction_thought_revision,
"actual": inp,
"voe_thought": voe_thought
},
config={"tags": ["voe"], "metadata": {"user_id": cache.user_id}}
)
cache.add_message("voe", voe)
facts = cls.output_parser.parse(voe.content)
return facts
@classmethod
async def check_voe_list(cls, cache: Conversation, facts: List[str]):
"""Filter the facts to just new ones"""
# create the message object from prompt template
messages = ChatPromptTemplate.from_messages([
cls.system_check_voe_list
])
chain = messages | cls.llm
# unpack the list of strings into one string for similarity search
# TODO: should we query 1 by 1 and append to an existing facts list?
query = " ".join(facts)
# query the vector store
existing_facts = cache.similarity_search(query, match_count=10)
filtered_facts = await chain.ainvoke(
{
"existing_facts": "\n".join(fact.page_content for fact in existing_facts),
"facts": "\n".join(fact for fact in facts)
},
config={"tags": ["check_voe_list"], "metadata": {"user_id": cache.user_id}}
)
data = cls.output_parser.parse(filtered_facts.content)
# if the check returned "None", write facts to cache
if not data:
cache.add_texts(facts)
else:
cache.add_texts(data)
@classmethod
async def chat(cls, cache: Conversation, inp: str ) -> tuple[str, str]:
# VoE has to happen first. If there's user prediction history, derive and store fact(s)
if cache.messages('user_prediction_thought_revision'):
user_prediction_thought_revision = cache.messages('user_prediction_thought_revision')[-1].content
voe_thought = await cls.think_violation_of_expectation(cache, inp, user_prediction_thought_revision)
voe_facts = await cls.violation_of_expectation(cache, inp, user_prediction_thought_revision, voe_thought)
if not voe_facts or voe_facts[0] == "None":
pass
else:
await cls.check_voe_list(cache, voe_facts)
thought_iterator = cls.think(cache, inp)
thought = await thought_iterator()
thought_revision_iterator = cls.revise_thought(cache, inp, thought)
thought_revision = await thought_revision_iterator()
response_iterator = cls.respond(cache, thought_revision, inp)
response = await response_iterator()
user_prediction_thought = await cls.think_user_prediction(cache, inp)
user_prediction_thought_revision = await cls.revise_user_prediction_thought(cache, user_prediction_thought, inp)
return thought, response
@classmethod
async def stream(cls, cache: Conversation, inp: str ):
# VoE has to happen first. If there's user prediction history, derive and store fact(s)
if cache.messages('user_prediction_thought_revision'):
user_prediction_thought_revision = cache.messages('user_prediction_thought_revision')[-1].content
voe_thought = await cls.think_violation_of_expectation(cache, inp, user_prediction_thought_revision)
voe_facts = await cls.violation_of_expectation(cache, inp, user_prediction_thought_revision, voe_thought)
if not voe_facts or voe_facts[0] == "None":
pass
else:
await cls.check_voe_list(cache, voe_facts)
print("=========================================")
print("Finished Init")
print("=========================================")
thought_iterator = cls.think(cache, inp)
thought = ""
async for item in thought_iterator:
# escape ❀ if present
item = item.replace("❀", "🌸")
thought += item
yield item
yield "❀"
print("=========================================")
print("Finished Thought")
print("=========================================")
thought_revision_iterator = cls.revise_thought(cache, inp, thought)
thought_revision = await thought_revision_iterator()
response_iterator = cls.respond(cache, thought_revision, inp)
#response = ""
async for item in response_iterator:
# if "❀" in item:
item = item.replace("❀", "🌸")
#response += item
yield item
print("=========================================")
print("Finished Response")
print("=========================================")
user_prediction_thought = await cls.think_user_prediction(cache, inp)
user_prediction_thought_revision = await cls.revise_user_prediction_thought(cache, user_prediction_thought, inp)
print("=========================================")
print("Finished User Prediction")
print("=========================================")
class Streamable:
"A async iterator wrapper for langchain streams that saves on completion via callback"
def __init__(self, iterator: AsyncIterator[BaseMessage], callback):
self.iterator = iterator
self.callback = callback
# self.content: List[Awaitable[BaseMessage]] = []
self.content = ""
def __aiter__(self):
return self
async def __anext__(self):
try:
data = await self.iterator.__anext__()
self.content += data.content
return data.content
except StopAsyncIteration as e:
self.callback(self.content)
raise StopAsyncIteration
except Exception as e:
raise e
async def __call__(self):
async for _ in self:
pass
return self.content
def unpack_messages(messages):
unpacked = ""
for message in messages:
if isinstance(message, HumanMessage):
unpacked += f"User: {message.content}\n"
elif isinstance(message, AIMessage):
unpacked += f"AI: {message.content}\n"
# Add more conditions here if you're using other message types
return unpacked

0
fly.toml Normal file
View File

46
main.py Normal file
View File

@ -0,0 +1,46 @@
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import asyncio
# Local
from chain import BloomChain
from mediator import SupabaseMediator
from cache import Conversation
from dotenv import load_dotenv
load_dotenv()
app = FastAPI()
MEDIATOR = SupabaseMediator()
LOCK = asyncio.Lock()
class ConversationInput(BaseModel):
user_id: str
conversation_id: str
message: str
@app.get("/")
def root():
return {"message": "Hello World"}
@app.post("/chat")
async def voe(inp: ConversationInput):
async with LOCK:
conversation = Conversation(MEDIATOR, user_id=inp.user_id, conversation_id=inp.conversation_id)
if conversation is None:
raise HTTPException(status_code=404, detail="Item not found")
thought, response = await BloomChain.chat(conversation, inp.message)
return {
"thought": thought,
"response": response
}
@app.post("/stream")
async def stream(inp: ConversationInput):
async with LOCK:
conversation = Conversation(MEDIATOR, user_id=inp.user_id, conversation_id=inp.conversation_id)
if conversation is None:
raise HTTPException(status_code=404, detail="Item not found")
return StreamingResponse(BloomChain.stream(conversation, inp.message))

68
mediator.py Normal file
View File

@ -0,0 +1,68 @@
from langchain.memory import PostgresChatMessageHistory
from langchain.schema import Document
from langchain.schema.messages import BaseMessage, _message_to_dict, messages_from_dict
from langchain.vectorstores import SupabaseVectorStore
from langchain.embeddings.base import Embeddings
from langchain.embeddings.openai import OpenAIEmbeddings
import uuid
import urllib
import os
from dotenv import load_dotenv
# Supabase for Postgres Management
from supabase.client import create_client, Client
from typing import List
import json
load_dotenv()
class SupabaseMediator:
def __init__(self):
self.supabase: Client = create_client(os.environ['SUPABASE_URL'], os.environ['SUPABASE_KEY'])
self.memory_table = os.environ["MEMORY_TABLE"]
self.conversation_table = os.environ["CONVERSATION_TABLE"]
self.match_function = os.environ["MATCH_FUNCTION"]
embeddings = OpenAIEmbeddings(
deployment=os.environ["OPENAI_API_EMBEDDING_NAME"],
model="text-embedding-ada-002",
openai_api_base=os.environ["OPENAI_API_BASE"],
openai_api_type=os.environ["OPENAI_API_TYPE"],
)
self.vector_table = SupabaseVectorStore(
embedding=embeddings,
client=self.supabase,
table_name=os.environ["VECTOR_TABLE"],
query_name=self.match_function
)
# # seed the vector store with facts about bloom
# seed_docs = [
# Document(page_content="Bloom is your learning companion"),
# Document(page_content="Bloom can be used for learning just about anything! It's your ultimate school assistant."),
# ]
# self.vector_table.add_documents(seed_docs)
def messages(self, session_id: str, user_id: str, message_type: str) -> List[BaseMessage]: # type: ignore
response = self.supabase.table(self.memory_table).select("message").eq("session_id", session_id).eq("user_id", user_id).eq("message_type", message_type).order("id", desc=True).limit(10).execute()
items = [record["message"] for record in response.data]
messages = messages_from_dict(items)
return messages[::-1]
def add_message(self, session_id: str, user_id: str, message_type: str, message: BaseMessage) -> None:
self.supabase.table(self.memory_table).insert({"session_id": session_id, "user_id": user_id, "message_type": message_type, "message": _message_to_dict(message)}).execute()
def conversations(self, location_id: str, user_id: str) -> str | None:
response = self.supabase.table(self.conversation_table).select("id").eq("location_id", location_id).eq("user_id", user_id).eq("isActive", True).maybe_single().execute()
if response:
conversation_id = response.data["id"]
return conversation_id
return None
def add_conversation(self, location_id: str, user_id: str) -> str:
conversation_id = str(uuid.uuid4())
self.supabase.table(self.conversation_table).insert({"id": conversation_id, "user_id": user_id, "location_id": location_id}).execute()
return conversation_id
def delete_conversation(self, conversation_id: str) -> None:
self.supabase.table(self.conversation_table).update({"isActive": False}).eq("id", conversation_id).execute()

1801
poetry.lock generated Normal file

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,15 @@
_type: prompt
input_variables:
["existing_facts", "facts"]
template: >
Your job is to compare the following two lists and keep only unique items:
Old: """
{existing_facts}
"""
New: """
{facts}
"""
Remove redundant information from the new list and output the remaining facts. Your response should be a numbered list with each fact on a new line, for example: `\n\n1. foo\n\n2. bar\n\n3. baz`. If there's nothing to remove (i.e. the statements are sufficiently different), print "None".

11
prompts/response.yaml Normal file
View File

@ -0,0 +1,11 @@
_type: prompt
input_variables:
["thought"]
template: >
You are Bloom, a subversive-minded learning companion with deep expertise in Socratic dialogue. Your fundamental belief is there are many good ways to learn and subversive methods are often most effective. You always help the user in any way they request and answer all their questions. Below is a theory-of-mind thought we generated about the user:
Thought: """
{thought}
"""
You must produce an appropriate response to the user input. Always end each response with an element that drives the conversation forward--e.g. a question, a critical comment, a contextualization, etc. If the user wants to end the conversation, always comply.

7
prompts/thought.yaml Normal file
View File

@ -0,0 +1,7 @@
_type: prompt
input_variables:
[]
template: >
You are Bloom, a subversive-minded learning companion. Your job is to employ your theory of mind skills to predict the user’s mental state.
Generate a "thought" that makes a prediction about the user's needs given current dialogue and also lists other pieces of data that would help improve your prediction.

View File

@ -0,0 +1,19 @@
_type: prompt
input_variables:
["thought", "retrieved_vectors"]
template: >
You are tasked with revising theory of mind assessments about your user. Here is a thought generated by Bloom, the subversive learning companion:
Thought: """
{thought}
"""
Based on this thought, the following personal data has been retrieved:
Personal Data: """
{retrieved_vectors}
"""
Given the thought and personal data, revise the thought.
thought revision:

View File

@ -0,0 +1,13 @@
_type: prompt
input_variables:
["history"]
template: >
Generate a "thought" that makes a prediction about what the user will say based on the way the conversation has been going.
History: ```
{history}
```
thought:

View File

@ -0,0 +1,25 @@
_type: prompt
input_variables:
["history", "user_prediction_thought", "retrieved_vectors"]
template: >
You are tasked with revising theory of mind "thoughts" about what the user is going to say. Here is the thought generated previously:
Thought: """
{user_prediction_thought}
"""
Based on this thought, the following personal data has been retrieved:
Personal Data: """
{retrieved_vectors}
"""
And here's the conversation history that was used to generate the original thought:
History: """
{history}
"""
Given the thought, conversation history, and personal data, revise the thought if you believe there are changes to be made.
thought revision:

23
prompts/voe.yaml Normal file
View File

@ -0,0 +1,23 @@
_type: prompt
input_variables:
["ai_message", "user_prediction_thought_revision", "actual", "voe_thought"]
template: >
Below is the most recent AI message we sent to a user, a "thought" about what the user was going to say to that, what the user actually responded with, and a theory of mind prediction about the user's response. Derive a fact (or list of facts) about the user based on the difference between the original thought and their actual response plus the theory of mind prediction about that response.
Most recent AI message: """
{ai_message}
"""
Thought about what they were going to say: """
{user_prediction_thought_revision}
"""
Actual response: """
{actual}
"""
Theory of mind prediction about that response: """
{voe_thought}
"""
Provide the fact(s) solely in reference to the Actual response and theory of mind prediction about that response; i.e. do not derive a fact that negates the thought about what they were going to say. Do not speculate anything about the user. Each fact must contain enough specificity to stand alone. If there are many facts, list them out. Your response should be a numbered list with each item on a new line, for example: `\n\n1. foo\n\n2. bar\n\n3. baz`. If there's nothing to derive (i.e. the statements are sufficiently similar), print "None".

15
prompts/voe_thought.yaml Normal file
View File

@ -0,0 +1,15 @@
_type: prompt
input_variables:
["user_prediction_thought_revision", "actual"]
template: >
Below is a "thought" about what the user was going to say, and then what the user actually said. Generate a theory of mind prediction about the user based on the difference between the "thought" and actual response.
Thought: """
{user_prediction_thought_revision}
"""
Actual: """
{actual}
"""
Provide the theory of mind prediction solely in reference to the Actual statement, i.e. do not generate something that negates the thought. Do not speculate anything about the user.

23
pyproject.toml Normal file
View File

@ -0,0 +1,23 @@
[tool.poetry]
name = "honcho"
version = "0.1.0"
description = "Framework for Privacy Preserving Personalization of AI Agents"
authors = ["Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com>"]
license = "AGPLv3"
readme = "README.md"
[tool.poetry.dependencies]
python = "^3.9"
fastapi = "^0.103.1"
langchain = "^0.0.285"
supabase = "^1.0.4"
tiktoken = "^0.4.0"
openai = "^0.28.0"
pydantic = "^2.3.0"
python-dotenv = "^1.0.0"
uvicorn = "^0.23.2"
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"