RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Anthropic Migration (#72) 

* ready to test deriver

* prompts ready for testing

* compare entire list, don't loop through facts one by one

* parse xml, batch process ai query docs

* timing works

* session id is writing to csv now

* ignore timing logs

* infinite backrooms-inpsired prompts

* yousim/ib inspired prompts v0

* accept incoming changes on agent.py

* dialectic to sonnet for now

* remove timing, use py-spy

* cleanup from timing removal

* remove extra timing stuff

* Route for User LeveL Metamessages

* functional state

* chore: Clean up artifacts

* fix(deriver) update database calls for new schema and payload

* chore(docs) Update Honcho Version

---------

Co-authored-by: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com>
This commit is contained in:
vintro 2024-10-17 23:39:23 -04:00 committed by GitHub
parent 6d5439a4d8
commit 487ec43362
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 9398 additions and 1239 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -175,3 +175,5 @@ cython_debug/
supabase/
docs/node_modules
timing_logs.csv

50
README.md
View File

@ -1,5 +1,6 @@
# 🫡 Honcho
![Static Badge](https://img.shields.io/badge/Version-0.0.11-blue)
![Static Badge](https://img.shields.io/badge/Version-0.0.12-blue)
[![Discord](https://img.shields.io/discord/1016845111637839922?style=flat&logo=discord&logoColor=23ffffff&label=Plastic%20Labs&labelColor=235865F2)](https://discord.gg/plasticlabs)
[![arXiv](https://img.shields.io/badge/arXiv-2310.06983-b31b1b.svg)](https://arxiv.org/abs/2310.06983)
![GitHub License](https://img.shields.io/github/license/plastic-labs/honcho)
@ -8,7 +9,6 @@
[![PyPI version](https://img.shields.io/pypi/v/honcho-ai.svg)](https://pypi.org/project/honcho-ai/)
[![NPM version](https://img.shields.io/npm/v/honcho-ai.svg)](https://npmjs.org/package/honcho-ai)
Honcho is a platform for making AI agents and LLM powered applications that are personalized
to their end users. It leverages the inherent theory-of-mind capabilities of
LLMs to cohere to user psychology over time.
@ -22,34 +22,34 @@ Read the user documentation [here](https://docs.honcho.dev)
- [Project Structure](#project-structure)
- [Usage](#usage)
- [Architecture](#architecture)
- [Storage](#storage)
- [Insights](#insights)
- [Storage](#storage)
- [Insights](#insights)
- [License](#license)
## Project Structure
The Honcho project is split between several repositories with this one hosting
the core service logic. This is implemented as a FastAPI server/API to store
data about an application's state.
data about an application's state.
There are also client-sdks that are created using
[Stainless](https://www.stainlessapi.com/). Currently, there is a [Python](https://github.com/plastic-labs/honcho-python) and
[TypeScript/JavaScript](https://github.com/plastic-labs/honcho-node) SDK available.
[Stainless](https://www.stainlessapi.com/). Currently, there is a [Python](https://github.com/plastic-labs/honcho-python) and
[TypeScript/JavaScript](https://github.com/plastic-labs/honcho-node) SDK available.
Examples on how to use the SDK are located within each SDK repository. There is
also SDK example usage available in the [API Reference](https://docs.honcho.dev/api-reference/introduction)
along with various guides.
along with various guides.
## Usage
Currently, there is a demo server of Honcho running at https://demo.honcho.dev.
This server is not production ready and does not have an reliability guarantees.
It is purely there for evaluation purposes.
It is purely there for evaluation purposes.
A private beta for a tenant isolated production ready version of Honcho is
currently underway. If interested fill out this
[typeform](https://plasticlabs.typeform.com/honchobeta) and the Plastic Labs
team will reach out to onboard users.
team will reach out to onboard users.
Additionally, Honcho can be self-hosted for testing and evaluation purposes. See
[Contributing](./CONTRIBUTING.md) for more details on how to setup a local
@ -58,18 +58,18 @@ version of Honcho.
## Architecture
The functionality of Honcho can be split into two different services: Storage
and Insights.
and Insights.
### Storage
### Storage
Honcho contains several different primitives used for storing application and
user data. This data is used for managing conversations, modeling user
psychology, building RAG applications, and more.
psychology, building RAG applications, and more.
The philosophy behind Honcho is to provide a platform that is user-centric and
easily scalable from a single user to a million.
easily scalable from a single user to a million.
Below is a mapping of the different primitives.
Below is a mapping of the different primitives.
```
Apps
@ -82,7 +82,7 @@ Apps
```
Users familiar with APIs such as the OpenAI Assistants API will be familiar with
much of the mapping here.
much of the mapping here.
#### Apps
@ -93,17 +93,17 @@ isolation data between use cases.
**Users**
Within an `App` everything revolves around a `User`. the `User` object
literally represent a user of an application.
literally represent a user of an application.
#### Sessions
The `Session` object represents a set of interactions a `User` has with an
`App`. Other application may refer to this as a thread or conversation.
`App`. Other application may refer to this as a thread or conversation.
**Messages**
The `Message` represents an atomic interaction of a `User` in a `Session`.
`Message`s are labed as either a `User` or AI message.
`Message`s are labed as either a `User` or AI message.
#### Metamessages
@ -112,32 +112,32 @@ meant to be used to store intermediate inference from AI assistants or other
derived information that is separate from the main `User` `App` interaction
loop. For complicated prompting architectures like [metacognitive prompting](https://arxiv.org/abs/2310.06983)
metamessages can store thought and reflection steps along with having developer
information such as logs.
information such as logs.
Each `Metamessage` is associated with a `Message`. The convention we recommend
is to attach a `Metamessage` to the `Message` it was derived from or based on.
is to attach a `Metamessage` to the `Message` it was derived from or based on.
#### Collections
At a high level a `Collection` is a named group of `Documents`. Developers
familiar with RAG based applications will be familar with these. `Collection`s
store vector embedded data that developers and agents can retrieve against using
functions like cosine similarity.
functions like cosine similarity.
Developers can create multiple `Collection`s for a user for different purposes
such as modeling different personas, adding third-party data such as emails and
PDF files, and more.
PDF files, and more.
#### Documents
As stated before a `Document` is vector embedded data stored in a `Collection`.
As stated before a `Document` is vector embedded data stored in a `Collection`.
### Insights
The Insight functionality of Honcho is built on top of the Storage service. As
`Messages` and `Sessions` are created for a `User`, Honcho will asynchronously
reason about the `User`'s psychology to derive facts about them and store them
in a reserved `Collection`.
in a reserved `Collection`.
To read more about how this works read our [Research Paper](https://arxiv.org/abs/2310.06983)

5287
docs/pnpm-lock.yaml Normal file
View File

File diff suppressed because it is too large Load Diff

2713
poetry.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff

2
pyproject.toml
View File

@ -23,8 +23,8 @@ dependencies = [
"opentelemetry-instrumentation-sqlalchemy>=0.45b0",
"opentelemetry-instrumentation-logging>=0.45b0",
"rich>=13.7.1",
"mirascope>=0.18.0",
"openai>=1.43.0",
"anthropic>=0.36.0",
"nanoid>=2.0.0",
]
[tool.uv]

170
src/agent.py
View File

@ -1,10 +1,13 @@
import asyncio
import os
from collections.abc import Iterable
from anthropic import Anthropic
from dotenv import load_dotenv
from mirascope.openai import OpenAICall, OpenAICallParams
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src import crud, schemas
from src import crud, models, schemas
from src.db import SessionLocal
load_dotenv()
@ -27,21 +30,56 @@ class AsyncSet:
return self._set.copy()
class Dialectic(OpenAICall):
prompt_template = """
You are tasked with responding to the query based on the context provided.
---
query: {agent_input}
context: {retrieved_facts}
conversation_history: {chat_history}
---
Provide a brief, matter-of-fact, and appropriate response to the query based on the context provided. If the context provided doesn't aid in addressing the query, return None.
"""
agent_input: str
retrieved_facts: str
chat_history: list[str]
class Dialectic:
def __init__(
self, agent_input: str, user_representation: str, chat_history: list[str]
):
self.agent_input = agent_input
self.user_representation = user_representation
self.chat_history = chat_history
self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
call_params = OpenAICallParams(model="gpt-4o", temperature=1.2, top_p=0.5)
def call(self):
prompt = f"""
You are tasked with responding to the query based on the context provided.
<query>{self.agent_input}</query>
<context>{self.user_representation}</context>
<conversation_history>{self.chat_history}</conversation_history>
Provide a brief, matter-of-fact, and appropriate response to the query based on the context provided. If the context provided doesn't aid in addressing the query, return only the word "None".
"""
response = self.client.messages.create(
messages=[
{
"role": "user",
"content": prompt,
}
],
model="claude-3-5-sonnet-20240620",
max_tokens=300,
)
return response.content
def stream(self):
prompt = f"""
You are tasked with responding to the query based on the context provided.
<query>{self.agent_input}</query>
<context>{self.user_representation}</context>
<conversation_history>{self.chat_history}</conversation_history>
Provide a brief, matter-of-fact, and appropriate response to the query based on the context provided. If the context provided doesn't aid in addressing the query, return only the word "None".
"""
yield from self.client.messages.create(
model="claude-3-5-sonnet-20240620",
messages=[
{
"role": "user",
"content": prompt,
}
],
max_tokens=300,
stream=True,
)
async def chat_history(app_id: str, user_id: str, session_id: str) -> list[str]:
@ -58,63 +96,27 @@ async def chat_history(app_id: str, user_id: str, session_id: str) -> list[str]:
return history
async def prep_inference(
app_id: str,
user_id: str,
query: str,
collection_name: str,
) -> None | list[str]:
async with SessionLocal() as db:
collection = await crud.get_collection_by_name(
db, app_id, user_id, collection_name
)
retrieved_facts = None
if collection:
retrieved_documents = await crud.query_documents(
db=db,
app_id=app_id,
user_id=user_id,
collection_id=collection.id,
query=query,
top_k=3,
)
if len(retrieved_documents) > 0:
retrieved_facts = [d.content for d in retrieved_documents]
return retrieved_facts
async def generate_facts(
app_id: str,
user_id: str,
fact_set: AsyncSet,
collection_name: str,
questions: list[str],
):
async def fetch_facts(query):
retrieved_facts = await prep_inference(app_id, user_id, query, collection_name)
if retrieved_facts is not None:
await fact_set.update(retrieved_facts)
await asyncio.gather(*[fetch_facts(query) for query in questions])
async def fact_generator(
app_id: str,
user_id: str,
collections: list[str],
questions: list[str],
):
fact_set = AsyncSet()
fact_tasks = [
generate_facts(app_id, user_id, fact_set, col, questions) for col in collections
]
await asyncio.gather(*fact_tasks)
fact_set_copy = fact_set.get_set()
facts = "None"
if fact_set_copy and len(fact_set_copy) > 0:
facts = "\n".join(fact_set_copy)
return facts
async def get_latest_user_representation(
db: AsyncSession, app_id: str, user_id: str, session_id: str
) -> str:
stmt = (
select(models.Metamessage)
.join(models.Message, models.Message.public_id == models.Metamessage.message_id)
.join(models.Session, models.Message.session_id == models.Session.public_id)
.join(models.User, models.User.public_id == models.Session.user_id)
.join(models.App, models.App.public_id == models.User.app_id)
.where(models.App.public_id == app_id)
.where(models.User.public_id == user_id)
.where(models.Metamessage.metamessage_type == "user_representation")
.limit(1)
)
result = await db.execute(stmt)
representation = result.scalar_one_or_none()
return (
representation.content
if representation
else "No user representation available."
)
async def chat(
@ -125,27 +127,23 @@ async def chat(
stream: bool = False,
):
questions = [query.queries] if isinstance(query.queries, str) else query.queries
final_query = "\n".join(questions) if len(questions) > 1 else questions[0]
collections = (
[query.collections] if isinstance(query.collections, str) else query.collections
)
async with SessionLocal() as db:
# Run user representation retrieval and chat history retrieval concurrently
user_rep_task = get_latest_user_representation(db, app_id, user_id, session_id)
history_task = chat_history(app_id, user_id, session_id)
# Run fact generation and chat history retrieval concurrently
fact_task = fact_generator(app_id, user_id, collections, questions)
history_task = chat_history(app_id, user_id, session_id)
# Wait for both tasks to complete
facts, history = await asyncio.gather(fact_task, history_task)
# Wait for both tasks to complete
user_representation, history = await asyncio.gather(user_rep_task, history_task)
chain = Dialectic(
agent_input=final_query,
retrieved_facts=facts,
user_representation=user_representation,
chat_history=history,
)
if stream:
return chain.stream_async()
return chain.stream()
response = chain.call()
return schemas.AgentChat(content=response.content)
return schemas.AgentChat(content=response[0].text)

9
src/crud.py
View File

@ -8,7 +8,6 @@ from sqlalchemy import Select, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
# from sqlalchemy.orm import Session
from . import models, schemas
load_dotenv(override=True)
@ -423,8 +422,8 @@ async def get_metamessages(
db: AsyncSession,
app_id: str,
user_id: str,
session_id: str,
message_id: Optional[str],
session_id: Optional[str] = None,
message_id: Optional[str] = None,
metamessage_type: Optional[str] = None,
filter: Optional[dict] = None,
reverse: Optional[bool] = False,
@ -437,9 +436,11 @@ async def get_metamessages(
.join(models.App, models.App.public_id == models.User.app_id)
.where(models.App.public_id == app_id)
.where(models.User.public_id == user_id)
.where(models.Message.session_id == session_id)
)
if session_id is not None:
stmt = stmt.where(models.Session.public_id == session_id)
if message_id is not None:
stmt = stmt.where(models.Metamessage.message_id == message_id)

3
src/deriver/__init__.py Normal file
View File

@ -0,0 +1,3 @@
from .consumer import process_item
__all__ = ['process_item']

3
src/deriver/__main__.py
View File

@ -1,4 +1,5 @@
import asyncio
import argparse
import uvloop
@ -6,4 +7,4 @@ from .queue import main
if __name__ == "__main__":
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main())
asyncio.run(main())

290
src/deriver/consumer.py
View File

@ -1,24 +1,14 @@
import logging
import re
from dotenv import load_dotenv
from rich import print as rprint
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from .. import crud, models, schemas
from ..db import SessionLocal
from .voe import (
CheckVoeList,
UserPredictionThought,
UserPredictionThoughtRevision,
VoeDeriveFacts,
VoeThought,
)
from .. import models
from .voe import tom_inference, user_representation
load_dotenv()
# Turn of SQLAlchemy Echo logging
# Turn off SQLAlchemy Echo logging
logging.getLogger("sqlalchemy.engine.Engine").disabled = True
@ -33,26 +23,18 @@ async def add_metamessage(db, message_id, metamessage_type, content):
db.add(metamessage)
def parse_xml_content(text, tag):
pattern = f"<{tag}>(.*?)</{tag}>"
match = re.search(pattern, text, re.DOTALL)
return match.group(1).strip() if match else ""
async def process_item(db: AsyncSession, payload: dict):
collection: models.Collection
collection = await crud.get_collection_by_name(
db, payload["app_id"], payload["user_id"], "honcho"
)
if collection is None:
collection_create = schemas.CollectionCreate(name="honcho", metadata={})
collection = await crud.create_collection(
db,
collection=collection_create,
app_id=payload["app_id"],
user_id=payload["user_id"],
)
collection_id = collection.id
processing_args = [
payload["content"],
payload["app_id"],
payload["user_id"],
payload["session_id"],
collection_id,
payload["message_id"],
db,
]
@ -68,7 +50,6 @@ async def process_ai_message(
app_id: str,
user_id: str,
session_id: str,
collection_id: str,
message_id: str,
db: AsyncSession,
):
@ -78,15 +59,15 @@ async def process_ai_message(
rprint(f"[green]Processing AI message: {content}[/green]")
subquery = (
select(models.Message.created_at)
.where(models.Message.id == message_id)
select(models.Message.id)
.where(models.Message.public_id == message_id)
.scalar_subquery()
)
messages_stmt = (
select(models.Message)
.where(models.Message.session_id == session_id)
.order_by(models.Message.created_at.desc())
.where(models.Message.created_at < subquery)
.order_by(models.Message.id.desc())
.where(models.Message.id < subquery)
.limit(10)
)
@ -99,101 +80,31 @@ async def process_ai_message(
# append current message to chat history
chat_history_str = f"{chat_history_str}\nai: {content}"
# user prediction thought
user_prediction_thought = UserPredictionThought(chat_history=chat_history_str)
user_prediction_thought_response = user_prediction_thought.call()
## query the collection to build the context
additional_data = re.findall(
r"\d+\.\s([^\n]+)", user_prediction_thought_response.content
tom_inference_response = await tom_inference(
chat_history_str, session_id=session_id
)
additional_data_set = set()
additional_data_list = []
for d in additional_data:
response = await crud.query_documents(
db,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
query=d,
top_k=3,
)
for document in response:
additional_data_set.add(document)
additional_data_list = list(additional_data_set)
context_str = "\n".join([document.content for document in additional_data_list])
prediction = parse_xml_content(tom_inference_response, "prediction")
# user prediction thought revision given the context
user_prediction_thought_revision = UserPredictionThoughtRevision(
user_prediction_thought=user_prediction_thought_response.content,
retrieved_context=context_str,
chat_history=chat_history_str,
await add_metamessage(
db,
message_id,
"tom_inference",
prediction,
)
user_prediction_thought_revision_response = user_prediction_thought_revision.call()
if user_prediction_thought_revision_response.content == "None":
rprint("[blue]Model predicted no changes to the user prediction thought")
await add_metamessage(
db,
message_id,
"user_prediction_thought",
user_prediction_thought_response.content,
)
await add_metamessage(
db,
message_id,
"user_prediction_thought_revision",
user_prediction_thought_response.content,
)
else:
await add_metamessage(
db,
message_id,
"user_prediction_thought",
user_prediction_thought_response.content,
)
await add_metamessage(
db,
message_id,
"user_prediction_thought_revision",
user_prediction_thought_revision_response.content,
)
await db.commit()
# debugging
rprint("[blue]=================")
rprint("[blue]User Prediction Thought Prompt:")
content_lines = str(user_prediction_thought)
rprint("[blue]Tom Inference:")
content_lines = str(prediction)
rprint(f"[blue]{content_lines}")
rprint("[blue]User Prediction Thought:")
content_lines = str(user_prediction_thought_response.content)
rprint(f"[blue]{content_lines}")
rprint("[blue]=================")
rprint("[medium_purple1]=================")
rprint("[medium_purple1]User Prediction Thought Revision:")
content_lines = str(user_prediction_thought_revision)
rprint(f"[medium_purple1]{content_lines}")
rprint("[medium_purple1]User Prediction Thought Revision Response:")
content_lines = str(user_prediction_thought_revision_response.content)
rprint(f"[medium_purple1]{content_lines}")
rprint("[medium_purple1]=================")
async def process_user_message(
content: str,
app_id: str,
user_id: str,
session_id: str,
collection_id: str,
message_id: str,
db: AsyncSession,
):
@ -201,11 +112,9 @@ async def process_user_message(
Process a user message. If there are revised user predictions to run VoE against, run it. Otherwise pass.
"""
rprint(f"[orange1]Processing User Message: {content}")
# Get the AI message directly preceding this User message
subquery = (
select(models.Message.created_at)
.where(models.Message.id == message_id)
select(models.Message.id)
.where(models.Message.public_id == message_id)
.scalar_subquery()
)
@ -213,8 +122,8 @@ async def process_user_message(
select(models.Message)
.where(models.Message.session_id == session_id)
.where(models.Message.is_user == False)
.order_by(models.Message.created_at.desc())
.where(models.Message.created_at < subquery)
.order_by(models.Message.id.desc())
.where(models.Message.id < subquery)
.limit(1)
)
@ -223,123 +132,62 @@ async def process_user_message(
if ai_message and ai_message.content:
rprint(f"[orange1]AI Message: {ai_message.content}")
# Get the User Thought Revision Associated with this AI Message
metamessages_stmt = (
# Fetch the tom_inference metamessage
tom_inference_stmt = (
select(models.Metamessage)
.where(models.Metamessage.message_id == ai_message.id)
.where(
models.Metamessage.metamessage_type
== "user_prediction_thought_revision"
)
.order_by(models.Metamessage.created_at.asc())
.where(models.Metamessage.message_id == ai_message.public_id)
.where(models.Metamessage.metamessage_type == "tom_inference")
.order_by(models.Metamessage.id.asc())
.limit(1)
)
response = await db.execute(metamessages_stmt)
metamessage = response.scalar_one_or_none()
response = await db.execute(tom_inference_stmt)
tom_inference_metamessage = response.scalar_one_or_none()
if metamessage and metamessage.content:
rprint(f"[orange1]Metamessage: {metamessage.content}")
if tom_inference_metamessage and tom_inference_metamessage.content:
rprint(f"[orange1]Tom Inference: {tom_inference_metamessage.content}")
# VoE thought
voe_thought = VoeThought(
user_prediction_thought_revision=metamessage.content, actual=content
# Fetch the existing user representation
user_representation_stmt = (
select(models.Metamessage)
.where(models.Metamessage.message_id == ai_message.public_id)
.where(models.Metamessage.metamessage_type == "user_representation")
.order_by(models.Metamessage.id.desc())
.limit(1)
)
voe_thought_response = voe_thought.call()
response = await db.execute(user_representation_stmt)
existing_representation = response.scalar_one_or_none()
# VoE derive facts
voe_derive_facts = VoeDeriveFacts(
ai_message=ai_message.content,
user_prediction_thought_revision=metamessage.content,
actual=content,
voe_thought=voe_thought_response.content,
existing_representation_content = (
existing_representation.content if existing_representation else "None"
)
voe_derive_facts_response = voe_derive_facts.call()
# debugging
rprint("[orange1]=================")
rprint("[orange1]Voe Thought Prompt:")
content_lines = str(voe_thought)
rprint(f"[orange1]{content_lines}")
rprint("[orange1]Voe Thought:")
content_lines = str(voe_thought_response.content)
rprint(f"[orange1]{content_lines}")
rprint("[orange1]=================")
# Call user_representation
user_representation_response = await user_representation(
chat_history=f"{ai_message.content}\nhuman: {content}",
session_id=session_id,
user_representation=existing_representation_content,
tom_inference=tom_inference_metamessage.content,
)
rprint("[orange1]================")
rprint("[orange1]Voe Derive Facts Prompt:")
content_lines = str(voe_derive_facts)
rprint(f"[orange1]{content_lines}")
# Store the user_representation response as a metamessage
await add_metamessage(
db,
message_id,
"user_representation",
user_representation_response,
)
rprint("[orange1]Voe Derive Facts Response:")
content_lines = str(voe_derive_facts_response.content)
rprint(f"[orange1]{content_lines}")
rprint("[orange1]=================")
# parse the user_representation response
user_representation_response = parse_xml_content(
user_representation_response, "representation"
)
facts = re.findall(r"\d+\.\s([^\n]+)", voe_derive_facts_response.content)
rprint("[orange1]=================")
rprint("[orange1]The Facts Themselves:")
rprint(facts)
new_facts = await check_dups(app_id, user_id, collection_id, facts)
rprint("[bright_magenta]User Representation:")
rprint(f"[bright_magenta]{user_representation_response}")
for fact in new_facts:
create_document = schemas.DocumentCreate(content=fact)
async with SessionLocal() as db:
doc = await crud.create_document(
db,
document=create_document,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
)
rprint(f"[orange1]Returned Document: {doc.content}")
else:
rprint("[red] No Prediction Associated with this Message")
return
raise Exception("\033[91mTom Inference NOT READY YET")
else:
rprint("[red]No AI message before this user message[/red]")
return
async def check_dups(app_id: str, user_id: str, collection_id: str, facts: list[str]):
"""Check that we're not storing duplicate facts"""
check_duplication = CheckVoeList(existing_facts=[], new_fact="")
result = None
new_facts = []
# global_existing_facts = [] # for debugging
for fact in facts:
async with SessionLocal() as db:
result = await crud.query_documents(
db=db,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
query=fact,
top_k=5,
)
existing_facts = [document.content for document in result]
if len(existing_facts) == 0:
new_facts.append(fact)
rprint(f"[light_steel_blue]New Fact: {fact}")
continue
# global_existing_facts.extend(existing_facts) # for debugging
check_duplication.existing_facts = existing_facts
check_duplication.new_fact = fact
response = check_duplication.call()
rprint("[light_steel_blue]==================")
rprint(f"[light_steel_blue]Dedupe Responses: {response.content}")
rprint("[light_steel_blue]==================")
if response.content == "true":
new_facts.append(fact)
rprint(f"[light_steel_blue]New Fact: {fact}")
continue
rprint("[light_steel_blue]===================")
# rprint("[light_steel_blue]Existing Facts:")
# rprint(global_existing_facts)
rprint("[light_steel_blue]Net New Facts:")
rprint(new_facts)
rprint("[light_steel_blue]===================")
return new_facts

7
src/deriver/queue.py
View File

@ -20,7 +20,7 @@ load_dotenv()
async def get_next_message_for_session(
db: AsyncSession, session_id: str
db: AsyncSession, session_id: int
) -> Optional[models.QueueItem]:
result = await db.execute(
select(models.QueueItem)
@ -33,7 +33,7 @@ async def get_next_message_for_session(
return result.scalar_one_or_none()
async def process_session_messages(session_id: str):
async def process_session_messages(session_id: int):
async with SessionLocal() as db:
try:
while True:
@ -110,6 +110,7 @@ async def schedule_session(
await db.commit()
# If successful, create a task for this session
# Pass enable_timing to process_session_messages
asyncio.create_task(process_session_messages(session_id))
except IntegrityError:
# If the session is already in active_sessions, skip it
@ -153,6 +154,6 @@ async def main():
AsyncioIntegration(),
],
)
semaphore = asyncio.Semaphore(6) # Limit to 5 concurrent dequeuing operations
semaphore = asyncio.Semaphore(2) # Limit to 5 concurrent dequeuing operations
queue_empty_flag = asyncio.Event() # Event to signal when the queue is empty
await polling_loop(semaphore, queue_empty_flag)

249
src/deriver/voe.py
View File

File diff suppressed because one or more lines are too long

15
src/main.py
View File

@ -198,7 +198,7 @@ app = FastAPI(
summary="An API for adding personalization to AI Apps",
description="""This API is used to store data and get insights about users for AI
applications""",
version="0.0.11",
version="0.0.12",
contact={
"name": "Plastic Labs",
"url": "https://plasticlabs.ai",
@ -233,11 +233,13 @@ add_pagination(app)
async def http_exception_handler(request, exc):
current_span = trace.get_current_span()
if (current_span is not None) and (current_span.is_recording()):
current_span.set_attributes({
"http.status_text": str(exc.detail),
"otel.status_description": f"{exc.status_code} / {str(exc.detail)}",
"otel.status_code": "ERROR",
})
current_span.set_attributes(
{
"http.status_text": str(exc.detail),
"otel.status_description": f"{exc.status_code} / {str(exc.detail)}",
"otel.status_code": "ERROR",
}
)
return PlainTextResponse(
json.dumps({"detail": str(exc.detail)}), status_code=exc.status_code
)
@ -248,5 +250,6 @@ app.include_router(users.router)
app.include_router(sessions.router)
app.include_router(messages.router)
app.include_router(metamessages.router)
app.include_router(metamessages.router_user_level)
app.include_router(collections.router)
app.include_router(documents.router)

2
src/models.py
View File

@ -58,7 +58,7 @@ class User(Base):
__table_args__ = (UniqueConstraint("name", "app_id", name="unique_name_app_user"),)
def __repr__(self) -> str:
return f"User(id={self.id}, app_id={self.app_id}, user_id={self.id}, created_at={self.created_at}, h_metadata={self.h_metadata})"
return f"User(id={self.id}, app_id={self.app_id}, created_at={self.created_at}, h_metadata={self.h_metadata})"
class Session(Base):

19
src/routers/apps.py
View File

@ -1,6 +1,6 @@
import traceback
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.exc import IntegrityError
from src import crud, schemas
@ -10,11 +10,12 @@ from src.security import auth
router = APIRouter(
prefix="/apps",
tags=["apps"],
dependencies=[Depends(auth)],
)
@router.get("/{app_id}", response_model=schemas.App)
async def get_app(request: Request, app_id: str, db=db, auth: dict = Depends(auth)):
async def get_app(app_id: str, db=db):
"""Get an App by ID
Args:
@ -31,9 +32,7 @@ async def get_app(request: Request, app_id: str, db=db, auth: dict = Depends(aut
@router.get("/name/{name}", response_model=schemas.App)
async def get_app_by_name(
request: Request, name: str, db=db, auth: dict = Depends(auth)
):
async def get_app_by_name(name: str, db=db):
"""Get an App by Name
Args:
@ -50,9 +49,7 @@ async def get_app_by_name(
@router.post("", response_model=schemas.App)
async def create_app(
request: Request, app: schemas.AppCreate, db=db, auth=Depends(auth)
):
async def create_app(app: schemas.AppCreate, db=db):
"""Create an App
Args:
@ -97,7 +94,7 @@ async def create_app(
@router.get("/get_or_create/{name}", response_model=schemas.App)
async def get_or_create_app(request: Request, name: str, db=db, auth=Depends(auth)):
async def get_or_create_app(name: str, db=db):
"""Get or Create an App
Args:
@ -110,17 +107,15 @@ async def get_or_create_app(request: Request, name: str, db=db, auth=Depends(aut
print("name", name)
app = await crud.get_app_by_name(db=db, name=name)
if app is None:
app = await create_app(request=request, db=db, app=schemas.AppCreate(name=name))
app = await create_app(db=db, app=schemas.AppCreate(name=name))
return app
@router.put("/{app_id}", response_model=schemas.App)
async def update_app(
request: Request,
app_id: str,
app: schemas.AppUpdate,
db=db,
auth=Depends(auth),
):
"""Update an App

15
src/routers/collections.py
View File

@ -1,7 +1,7 @@
import json
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi import APIRouter, Depends, HTTPException
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
@ -12,18 +12,17 @@ from src.security import auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/collections",
tags=["collections"],
dependencies=[Depends(auth)],
)
@router.get("", response_model=Page[schemas.Collection])
async def get_collections(
request: Request,
app_id: str,
user_id: str,
reverse: Optional[bool] = False,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
"""Get All Collections for a User
@ -49,12 +48,10 @@ async def get_collections(
@router.get("/name/{name}", response_model=schemas.Collection)
async def get_collection_by_name(
request: Request,
app_id: str,
user_id: str,
name: str,
db=db,
auth=Depends(auth),
) -> schemas.Collection:
honcho_collection = await crud.get_collection_by_name(
db, app_id=app_id, user_id=user_id, name=name
@ -68,12 +65,10 @@ async def get_collection_by_name(
@router.get("/{collection_id}", response_model=schemas.Collection)
async def get_collection_by_id(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
db=db,
auth=Depends(auth),
) -> schemas.Collection:
honcho_collection = await crud.get_collection_by_id(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
@ -87,12 +82,10 @@ async def get_collection_by_id(
@router.post("", response_model=schemas.Collection)
async def create_collection(
request: Request,
app_id: str,
user_id: str,
collection: schemas.CollectionCreate,
db=db,
auth=Depends(auth),
):
if collection.name == "honcho":
raise HTTPException(
@ -112,13 +105,11 @@ async def create_collection(
@router.put("/{collection_id}", response_model=schemas.Collection)
async def update_collection(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
collection: schemas.CollectionUpdate,
db=db,
auth=Depends(auth),
):
if collection.name is None and collection.metadata is None:
raise HTTPException(
@ -148,12 +139,10 @@ async def update_collection(
@router.delete("/{collection_id}")
async def delete_collection(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
db=db,
auth=Depends(auth),
):
try:
await crud.delete_collection(

13
src/routers/documents.py
View File

@ -12,19 +12,18 @@ from src.security import auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/collections/{collection_id}/documents",
tags=["documents"],
dependencies=[Depends(auth)],
)
@router.get("", response_model=Page[schemas.Document])
async def get_documents(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
reverse: Optional[bool] = False,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
try:
data = None
@ -54,13 +53,11 @@ async def get_documents(
response_model=schemas.Document,
)
async def get_document(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
document_id: str,
db=db,
auth=Depends(auth),
):
honcho_document = await crud.get_document(
db,
@ -78,7 +75,6 @@ async def get_document(
@router.get("/query", response_model=Sequence[schemas.Document])
async def query_documents(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
@ -86,7 +82,6 @@ async def query_documents(
top_k: int = 5,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
if top_k is not None and top_k > 50:
top_k = 50 # TODO see if we need to paginate this
@ -106,13 +101,11 @@ async def query_documents(
@router.post("", response_model=schemas.Document)
async def create_document(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
document: schemas.DocumentCreate,
db=db,
auth=Depends(auth),
):
try:
return await crud.create_document(
@ -133,14 +126,12 @@ async def create_document(
response_model=schemas.Document,
)
async def update_document(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
document_id: str,
document: schemas.DocumentUpdate,
db=db,
auth=Depends(auth),
):
if document.content is None and document.metadata is None:
raise HTTPException(
@ -163,13 +154,11 @@ async def update_document(
@router.delete("/{document_id}")
async def delete_document(
request: Request,
app_id: str,
user_id: str,
collection_id: str,
document_id: str,
db=db,
auth=Depends(auth),
):
response = await crud.delete_document(
db,

13
src/routers/messages.py
View File

@ -1,7 +1,7 @@
import json
from typing import Optional
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
@ -14,6 +14,7 @@ from src.security import auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions/{session_id}/messages",
tags=["messages"],
dependencies=[Depends(auth)],
)
@ -56,14 +57,12 @@ async def enqueue(payload: dict):
@router.post("", response_model=schemas.Message)
async def create_message_for_session(
request: Request,
app_id: str,
user_id: str,
session_id: str,
message: schemas.MessageCreate,
background_tasks: BackgroundTasks,
db=db,
auth=Depends(auth),
):
"""Adds a message to a session
@ -88,7 +87,7 @@ async def create_message_for_session(
"app_id": app_id,
"user_id": user_id,
"session_id": session_id,
"message_id": honcho_message.id,
"message_id": honcho_message.public_id,
"is_user": honcho_message.is_user,
"content": honcho_message.content,
"metadata": honcho_message.h_metadata,
@ -105,14 +104,12 @@ async def create_message_for_session(
@router.get("", response_model=Page[schemas.Message])
async def get_messages(
request: Request,
app_id: str,
user_id: str,
session_id: str,
reverse: Optional[bool] = False,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
"""Get all messages for a session
@ -151,13 +148,11 @@ async def get_messages(
@router.get("/{message_id}", response_model=schemas.Message)
async def get_message(
request: Request,
app_id: str,
user_id: str,
session_id: str,
message_id: str,
db=db,
auth=Depends(auth),
):
""" """
honcho_message = await crud.get_message(
@ -170,14 +165,12 @@ async def get_message(
@router.put("/{message_id}", response_model=schemas.Message)
async def update_message(
request: Request,
app_id: str,
user_id: str,
session_id: str,
message_id: str,
message: schemas.MessageUpdate,
db=db,
auth=Depends(auth),
):
"""Update's the metadata of a message"""
if message.metadata is None:

59
src/routers/metamessages.py
View File

@ -11,19 +11,24 @@ from src.security import auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions/{session_id}/metamessages",
tags=["messages"],
tags=["metamessages"],
dependencies=[Depends(auth)],
)
router_user_level = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/metamessages",
tags=["metamessages"],
dependencies=[Depends(auth)],
)
@router.post("", response_model=schemas.Metamessage)
async def create_metamessage(
request: Request,
app_id: str,
user_id: str,
session_id: str,
metamessage: schemas.MetamessageCreate,
db=db,
auth=Depends(auth),
):
"""Adds a message to a session
@ -55,7 +60,6 @@ async def create_metamessage(
@router.get("", response_model=Page[schemas.Metamessage])
async def get_metamessages(
request: Request,
app_id: str,
user_id: str,
session_id: str,
@ -64,7 +68,6 @@ async def get_metamessages(
reverse: Optional[bool] = False,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
"""Get all messages for a session
@ -103,19 +106,59 @@ async def get_metamessages(
raise HTTPException(status_code=404, detail="Session not found") from None
@router_user_level.get("", response_model=Page[schemas.Metamessage])
async def get_metamessages_by_user(
app_id: str,
user_id: str,
metamessage_type: Optional[str] = None,
reverse: Optional[bool] = False,
filter: Optional[str] = None,
db=db,
):
"""Paginate through the user metamessages for a user
Args:
app_id (str): The ID of the app representing the client application using honcho
user_id (str): The User ID representing the user, managed by the user
reverse (bool): Whether to reverse the order of the metamessages
Returns:
list[schemas.Message]: List of Message objects
Raises:
HTTPException: If the session is not found
"""
try:
data = None
if filter is not None:
data = json.loads(filter)
return await paginate(
db,
await crud.get_metamessages(
db,
app_id=app_id,
user_id=user_id,
metamessage_type=metamessage_type,
reverse=reverse,
filter=data,
),
)
except ValueError:
raise HTTPException(status_code=404, detail="User not found") from None
@router.get(
"/{metamessage_id}",
response_model=schemas.Metamessage,
)
async def get_metamessage(
request: Request,
app_id: str,
user_id: str,
session_id: str,
message_id: str,
metamessage_id: str,
db=db,
auth=Depends(auth),
):
"""Get a specific Metamessage by ID
@ -149,14 +192,12 @@ async def get_metamessage(
response_model=schemas.Metamessage,
)
async def update_metamessage(
request: Request,
app_id: str,
user_id: str,
session_id: str,
metamessage_id: str,
metamessage: schemas.MetamessageUpdate,
db=db,
auth=Depends(auth),
):
"""Update's the metadata of a metamessage"""
if metamessage.metadata is None:

10
src/routers/sessions.py
View File

@ -13,6 +13,7 @@ from src.security import auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions",
tags=["sessions"],
dependencies=[Depends(auth)],
)
@ -24,7 +25,6 @@ async def get_sessions(
reverse: Optional[bool] = False,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
"""Get All Sessions for a User
@ -61,7 +61,6 @@ async def create_session(
user_id: str,
session: schemas.SessionCreate,
db=db,
auth=Depends(auth),
):
"""Create a Session for a User
@ -95,7 +94,6 @@ async def update_session(
session_id: str,
session: schemas.SessionUpdate,
db=db,
auth=Depends(auth),
):
"""Update the metadata of a Session
@ -126,7 +124,6 @@ async def delete_session(
user_id: str,
session_id: str,
db=db,
auth=Depends(auth),
):
"""Delete a session by marking it as inactive
@ -158,7 +155,6 @@ async def get_session(
user_id: str,
session_id: str,
db=db,
auth=Depends(auth),
):
"""Get a specific session for a user by ID
@ -188,7 +184,6 @@ async def chat(
user_id: str,
session_id: str,
query: schemas.AgentQuery,
auth=Depends(auth),
):
print(query)
return await agent.chat(
@ -212,7 +207,6 @@ async def get_chat_stream(
user_id: str,
session_id: str,
query: schemas.AgentQuery,
auth=Depends(auth),
):
async def parse_stream():
stream = await agent.chat(
@ -222,7 +216,7 @@ async def get_chat_stream(
query=query,
stream=True,
)
async for chunk in stream:
for chunk in stream:
yield chunk.content
return StreamingResponse(

19
src/routers/users.py
View File

@ -1,7 +1,7 @@
import json
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi import APIRouter, Depends, HTTPException
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from sqlalchemy.exc import IntegrityError
@ -13,16 +13,15 @@ from src.security import auth
router = APIRouter(
prefix="/apps/{app_id}/users",
tags=["users"],
dependencies=[Depends(auth)],
)
@router.post("", response_model=schemas.User)
async def create_user(
request: Request,
app_id: str,
user: schemas.UserCreate,
db=db,
auth=Depends(auth),
):
"""Create a User
@ -46,12 +45,10 @@ async def create_user(
@router.get("", response_model=Page[schemas.User])
async def get_users(
request: Request,
app_id: str,
reverse: bool = False,
filter: Optional[str] = None,
db=db,
auth=Depends(auth),
):
"""Get All Users for an App
@ -74,11 +71,9 @@ async def get_users(
@router.get("/name/{name}", response_model=schemas.User)
async def get_user_by_name(
request: Request,
app_id: str,
name: str,
db=db,
auth=Depends(auth),
):
"""Get a User
@ -99,11 +94,9 @@ async def get_user_by_name(
@router.get("/{user_id}", response_model=schemas.User)
async def get_user(
request: Request,
app_id: str,
user_id: str,
db=db,
auth=Depends(auth),
):
"""Get a User
@ -123,9 +116,7 @@ async def get_user(
@router.get("/get_or_create/{name}", response_model=schemas.User)
async def get_or_create_user(
request: Request, app_id: str, name: str, db=db, auth=Depends(auth)
):
async def get_or_create_user(app_id: str, name: str, db=db):
"""Get or Create a User
Args:
@ -140,19 +131,17 @@ async def get_or_create_user(
user = await crud.get_user_by_name(db, app_id=app_id, name=name)
if user is None:
user = await create_user(
request=request, db=db, app_id=app_id, user=schemas.UserCreate(name=name)
db=db, app_id=app_id, user=schemas.UserCreate(name=name)
)
return user
@router.put("/{user_id}", response_model=schemas.User)
async def update_user(
request: Request,
app_id: str,
user_id: str,
user: schemas.UserUpdate,
db=db,
auth=Depends(auth),
):
"""Update a User

2
src/schemas.py
View File

@ -268,7 +268,7 @@ class Document(DocumentBase):
class AgentQuery(BaseModel):
queries: str | list[str]
collections: str | list[str] = "honcho"
# collections: str | list[str] = "honcho"
class AgentChat(BaseModel):

129
tests/routes/test_metamessages.py
View File

@ -65,6 +65,135 @@ async def test_get_metamessage(client, db_session, sample_data):
assert data["metamessage_type"] == "test_type"
@pytest.mark.asyncio
async def test_get_metamessages(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.public_id)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.public_id, content="Test message", is_user=True
)
db_session.add(test_message)
await db_session.commit()
test_metamessage_1 = models.Metamessage(
message_id=test_message.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type",
)
test_metamessage_2 = models.Metamessage(
message_id=test_message.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type",
)
test_metamessage_3 = models.Metamessage(
message_id=test_message.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type",
)
test_metamessage_4 = models.Metamessage(
message_id=test_message.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type_2",
)
db_session.add(test_metamessage_1)
db_session.add(test_metamessage_2)
db_session.add(test_metamessage_3)
db_session.add(test_metamessage_4)
await db_session.commit()
response = client.get(
f"/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/metamessages?metamessage_type=test_type"
)
assert response.status_code == 200
data = response.json()
assert "items" in data
assert len(data["items"]) > 0
assert len(data["items"]) == 3
assert data["items"][0]["content"] == "Test Metamessage"
assert data["items"][0]["metamessage_type"] == "test_type"
assert data["items"][0]["metadata"] == {}
@pytest.mark.asyncio
async def test_get_metamessage_by_user(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a 3 test sessions
test_session_1 = models.Session(user_id=test_user.public_id)
test_session_2 = models.Session(user_id=test_user.public_id)
test_session_3 = models.Session(user_id=test_user.public_id)
db_session.add(test_session_1)
db_session.add(test_session_2)
db_session.add(test_session_3)
await db_session.commit()
# Create a message in each session
test_message_1 = models.Message(
session_id=test_session_1.public_id, content="Test message", is_user=True
)
test_message_2 = models.Message(
session_id=test_session_2.public_id, content="Test message", is_user=True
)
test_message_3 = models.Message(
session_id=test_session_3.public_id, content="Test message", is_user=True
)
db_session.add(test_message_1)
db_session.add(test_message_2)
db_session.add(test_message_3)
await db_session.commit()
# create a metamessage on each message
test_metamessage_1 = models.Metamessage(
message_id=test_message_1.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type",
)
test_metamessage_2 = models.Metamessage(
message_id=test_message_2.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type",
)
test_metamessage_3 = models.Metamessage(
message_id=test_message_3.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type",
)
test_metamessage_4 = models.Metamessage(
message_id=test_message_3.public_id,
content="Test Metamessage",
metadata={},
metamessage_type="test_type_2",
)
db_session.add(test_metamessage_1)
db_session.add(test_metamessage_2)
db_session.add(test_metamessage_3)
db_session.add(test_metamessage_4)
await db_session.commit()
response = client.get(
f"/apps/{test_app.public_id}/users/{test_user.public_id}/metamessages?metamessage_type=test_type"
)
assert response.status_code == 200
data = response.json()
print("=======")
print(data)
print("=======")
assert len(data["items"]) > 0
assert len(data["items"]) == 3
assert data["items"][0]["content"] == "Test Metamessage"
assert data["items"][0]["metamessage_type"] == "test_type"
assert data["items"][0]["metadata"] == {}
@pytest.mark.asyncio
async def test_update_metamessage(client, db_session, sample_data):
test_app, test_user = sample_data

1556
uv.lock
View File

File diff suppressed because it is too large Load Diff