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

Dialectic Endpoint Improvements (#67) 

* feat(dialectic) Allow for batch questions and load session history

* feat(dialectic) parallelize facts and history queries

* feat(agent) Addresses dev-258 allow specifying additional collections

* feat(uv) switched from poetry to uv

* feat(deriver) Turn off derivations by editing session medatadata with a deriver_disabled flag

* fix(tests) Clean up test logic

---------

Co-authored-by: Vineeth Voruganti <vineeth@macbook-pro.mynetworksettings.com>
This commit is contained in:
Vineeth Voruganti 2024-09-14 17:06:11 -04:00 committed by GitHub
parent 5cac1184e5
commit 25229921c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
16 changed files with 2120 additions and 2618 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.dockerignore
View File

@ -11,4 +11,4 @@ docker-compose.yml.example
.github/**
.vscode/**
data/**
.venv

1
.gitignore vendored
View File

@ -5,6 +5,7 @@ api/docker-compose.yml
*.db
data
docker-compose.yml
compose.yml
# Byte-compiled / optimized / DLL files

48
Dockerfile
View File

@ -2,39 +2,41 @@
# https://testdriven.io/blog/docker-best-practices/
FROM python:3.11-slim-bullseye
RUN apt-get update && apt-get install -y build-essential
WORKDIR /app
# https://stackoverflow.com/questions/53835198/integrating-python-poetry-with-docker
ENV PYTHONFAULTHANDLER=1 \
PYTHONUNBUFFERED=1 \
PYTHONHASHSEED=random \
PIP_NO_CACHE_DIR=off \
PIP_DISABLE_PIP_VERSION_CHECK=on \
PIP_DEFAULT_TIMEOUT=100 \
POETRY_VERSION=1.8.3
RUN pip install "poetry==$POETRY_VERSION"
# Copy only requirements to cache them in docker layer
WORKDIR /app
COPY poetry.lock pyproject.toml /app/
# Project initialization:
RUN poetry config virtualenvs.create false \
&& poetry install --no-root --no-interaction --no-ansi
COPY --from=ghcr.io/astral-sh/uv:0.4.9 /uv /bin/uv
# Set Working directory
WORKDIR /app
RUN addgroup --system app && adduser --system --group app
RUN chown -R app:app /app
USER app
# Enable bytecode compilation
ENV UV_COMPILE_BYTECODE=1
# Copy from the cache instead of linking since it's a mounted volume
ENV UV_LINK_MODE=copy
# Install the project's dependencies using the lockfile and settings
RUN --mount=type=cache,target=/root/.cache/uv \
--mount=type=bind,source=uv.lock,target=uv.lock \
--mount=type=bind,source=pyproject.toml,target=pyproject.toml \
uv sync --frozen --no-install-project --no-dev
# Copy only requirements to cache them in docker layer
COPY uv.lock pyproject.toml /app/
# Sync the project
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --frozen --no-dev
# Place executables in the environment at the front of the path
ENV PATH="/app/.venv/bin:$PATH"
COPY --chown=app:app src/ /app/src/
EXPOSE 8000
# https://stackoverflow.com/questions/29663459/python-app-does-not-print-anything-when-running-detached-in-docker
CMD ["fastapi", "run", "src/main.py"]
CMD ["fastapi", "dev", "--host", "0.0.0.0", "src/main.py"]

3
docker-compose.yml.example
View File

@ -1,4 +1,3 @@
version: "3.8"
services:
api:
image: honcho:latest
@ -18,7 +17,7 @@ services:
build:
context: .
dockerfile: Dockerfile
entrypoint: ["python", "-m", "src.deriver"]
entrypoint: ["uv", "run", "python", "-m", "src.deriver"]
depends_on:
database:
condition: service_healthy

2475
poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

69
pyproject.toml
View File

@ -1,37 +1,39 @@
[tool.poetry]
[project]
name = "honcho"
version = "0.0.11"
version = "0.0.12"
description = "Honcho Server"
authors = ["Plastic Labs <hello@plasticlabs.ai>"]
authors = [
{name = "Plastic Labs", email = "hello@plasticlabs.ai"},
]
readme = "README.md"
package-mode = false
[tool.poetry.dependencies]
python = "^3.9"
fastapi = "^0.111.0"
python-dotenv = "^1.0.0"
sqlalchemy = "^2.0.30"
fastapi-pagination = "^0.12.24"
pgvector = "^0.2.5"
sentry-sdk = {extras = ["fastapi", "sqlalchemy"], version = "^2.3.1"}
greenlet = "^3.0.3"
psycopg = {extras= ["binary"], version="^3.1.19"}
httpx = "^0.27.0"
opentelemetry-instrumentation-fastapi = "^0.45b0"
opentelemetry-sdk = "^1.24.0"
opentelemetry-exporter-otlp = "^1.24.0"
opentelemetry-instrumentation-sqlalchemy = "^0.45b0"
opentelemetry-instrumentation-logging = "^0.45b0"
rich = "^13.7.1"
mirascope = "^0.18.0"
openai = "^1.43.0"
[tool.poetry.group.test.dependencies]
pytest = "^8.2.2"
sqlalchemy-utils = "^0.41.2"
pytest-asyncio = "^0.23.7"
coverage = "^7.6.0"
interrogate = "^1.7.0"
requires-python = ">=3.9"
dependencies = [
"fastapi[standard]>=0.111.0",
"python-dotenv>=1.0.0",
"sqlalchemy>=2.0.30",
"fastapi-pagination>=0.12.24",
"pgvector>=0.2.5",
"sentry-sdk[fastapi,sqlalchemy]>=2.3.1",
"greenlet>=3.0.3",
"psycopg[binary]>=3.1.19",
"httpx>=0.27.0",
"opentelemetry-instrumentation-fastapi>=0.45b0",
"opentelemetry-sdk>=1.24.0",
"opentelemetry-exporter-otlp>=1.24.0",
"opentelemetry-instrumentation-sqlalchemy>=0.45b0",
"opentelemetry-instrumentation-logging>=0.45b0",
"rich>=13.7.1",
"mirascope>=0.18.0",
"openai>=1.43.0",
]
[tool.uv]
dev-dependencies = [
"pytest>=8.2.2",
"sqlalchemy-utils>=0.41.2",
"pytest-asyncio>=0.23.7",
"coverage>=7.6.0",
"interrogate>=1.7.0",
]
[tool.ruff.lint]
# from https://docs.astral.sh/ruff/linter/#rule-selection example
@ -54,10 +56,5 @@ ignore = ["E501"]
[tool.ruff.flake8-bugbear]
extend-immutable-calls = ["fastapi.Depends"]
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"
[tool.lpytest.ini_options]
asyncio_mode = "auto"

160
src/agent.py
View File

@ -1,27 +1,48 @@
import asyncio
import os
import uuid
from typing import Iterable, Set
from dotenv import load_dotenv
from mirascope.base import BaseConfig
from mirascope.openai import OpenAICall, OpenAICallParams, azure_client_wrapper
from sqlalchemy.ext.asyncio import AsyncSession
from . import crud, schemas
from src import crud, schemas
from src.db import SessionLocal
load_dotenv()
class AsyncSet:
def __init__(self):
self._set: Set[str] = set()
self._lock = asyncio.Lock()
async def add(self, item: str):
async with self._lock:
self._set.add(item)
async def update(self, items: Iterable[str]):
async with self._lock:
self._set.update(items)
def get_set(self) -> Set[str]:
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]
configuration = BaseConfig(
client_wrappers=[
@ -35,61 +56,112 @@ class Dialectic(OpenAICall):
call_params = OpenAICallParams(
model=os.getenv("AZURE_OPENAI_DEPLOYMENT"), temperature=1.2, top_p=0.5
)
# call_params = OpenAICallParams(model="gpt-4o-2024-05-13")
async def chat_history(
app_id: uuid.UUID, user_id: uuid.UUID, session_id: uuid.UUID
) -> list[str]:
async with SessionLocal() as db:
stmt = await crud.get_messages(db, app_id, user_id, session_id)
results = await db.execute(stmt)
messages = results.scalars()
history = []
for message in messages:
if message.is_user:
history.append(f"user:{message.content}")
else:
history.append(f"assistant:{message.content}")
return history
async def prep_inference(
db: AsyncSession,
app_id: uuid.UUID,
user_id: uuid.UUID,
query: str,
):
collection = await crud.get_collection_by_name(db, app_id, user_id, "honcho")
retrieved_facts = None
if collection is None:
collection_create = schemas.CollectionCreate(name="honcho", metadata={})
collection = await crud.create_collection(
db,
collection=collection_create,
app_id=app_id,
user_id=user_id,
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
)
else:
retrieved_documents = await crud.query_documents(
db=db,
app_id=app_id,
user_id=user_id,
collection_id=collection.id,
query=query,
top_k=1,
)
if len(retrieved_documents) > 0:
retrieved_facts = retrieved_documents[0].content
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]
chain = Dialectic(
agent_input=query,
retrieved_facts=retrieved_facts if retrieved_facts else "None",
)
return chain
return retrieved_facts
async def generate_facts(
app_id: uuid.UUID,
user_id: uuid.UUID,
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: uuid.UUID,
user_id: uuid.UUID,
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 chat(
app_id: uuid.UUID,
user_id: uuid.UUID,
query: str,
db: AsyncSession,
session_id: uuid.UUID,
query: schemas.AgentQuery,
stream: bool = False,
):
chain = await prep_inference(db, app_id, user_id, query)
response = await chain.call_async()
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
)
# 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)
chain = Dialectic(
agent_input=final_query,
retrieved_facts=facts,
chat_history=history,
)
if stream:
return chain.stream_async()
response = chain.call()
return schemas.AgentChat(content=response.content)
async def stream(
app_id: uuid.UUID,
user_id: uuid.UUID,
query: str,
db: AsyncSession,
):
chain = await prep_inference(db, app_id, user_id, query)
return chain.stream_async()

12
src/deriver/consumer.py
View File

@ -95,9 +95,9 @@ async def process_ai_message(
result = await db.execute(messages_stmt)
messages = result.scalars().all()[::-1]
chat_history_str = "\n".join([
f"human: {m.content}" if m.is_user else f"ai: {m.content}" for m in messages
])
chat_history_str = "\n".join(
[f"human: {m.content}" if m.is_user else f"ai: {m.content}" for m in messages]
)
# append current message to chat history
chat_history_str = f"{chat_history_str}\nai: {content}"
@ -203,6 +203,8 @@ 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)
@ -223,6 +225,7 @@ 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 = (
select(models.Metamessage)
.where(models.Metamessage.message_id == ai_message.id)
@ -292,7 +295,8 @@ async def process_user_message(
)
rprint(f"[orange1]Returned Document: {doc.content}")
else:
raise Exception("\033[91mUser Thought Prediction Revision NOT READY YET")
rprint("[red] No Prediction Associated with this Message")
return
else:
rprint("[red]No AI message before this user message[/red]")
return

2
src/deriver/queue.py
View File

@ -94,7 +94,7 @@ async def schedule_session(
async with semaphore, SessionLocal() as db:
try:
available_slots = semaphore._value
print(available_slots)
# print(available_slots)
new_sessions = await get_available_sessions(db, available_slots)
if new_sessions:

21
src/routers/messages.py
View File

@ -20,6 +20,25 @@ router = APIRouter(
async def enqueue(payload: dict):
async with SessionLocal() as db:
# Get Session and Check metadata
session = await crud.get_session(
db,
app_id=payload["app_id"],
user_id=payload["user_id"],
session_id=payload["session_id"],
)
# Check if metadata has a "deriver" key
if session is not None:
deriver_disabled = session.h_metadata.get("deriver_disabled")
if deriver_disabled is not None and deriver_disabled is not False:
print("=====================")
print(f"Deriver is not enabled on session {payload['session_id']}")
print("=====================")
# If deriver is not enabled, do not enqueue
return
else:
# Session doesn't exist return
return
try:
processed_payload = {
k: str(v) if isinstance(v, uuid.UUID) else v for k, v in payload.items()
@ -68,8 +87,6 @@ async def create_message_for_session(
honcho_message = await crud.create_message(
db, message=message, app_id=app_id, user_id=user_id, session_id=session_id
)
print("=======")
print("Should be enqueued")
payload = {
"app_id": app_id,
"user_id": user_id,

34
src/routers/sessions.py
View File

@ -2,7 +2,7 @@ import json
import uuid
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
@ -19,7 +19,6 @@ router = APIRouter(
@router.get("", response_model=Page[schemas.Session])
async def get_sessions(
request: Request,
app_id: uuid.UUID,
user_id: uuid.UUID,
is_active: Optional[bool] = False,
@ -59,7 +58,6 @@ async def get_sessions(
@router.post("", response_model=schemas.Session)
async def create_session(
request: Request,
app_id: uuid.UUID,
user_id: uuid.UUID,
session: schemas.SessionCreate,
@ -93,7 +91,6 @@ async def create_session(
@router.put("/{session_id}", response_model=schemas.Session)
async def update_session(
request: Request,
app_id: uuid.UUID,
user_id: uuid.UUID,
session_id: uuid.UUID,
@ -126,7 +123,6 @@ async def update_session(
@router.delete("/{session_id}")
async def delete_session(
request: Request,
app_id: uuid.UUID,
user_id: uuid.UUID,
session_id: uuid.UUID,
@ -159,7 +155,6 @@ async def delete_session(
@router.get("/{session_id}", response_model=schemas.Session)
async def get_session(
request: Request,
app_id: uuid.UUID,
user_id: uuid.UUID,
session_id: uuid.UUID,
@ -188,20 +183,21 @@ async def get_session(
return honcho_session
@router.get("/{session_id}/chat", response_model=schemas.AgentChat)
async def get_chat(
request: Request,
@router.post("/{session_id}/chat", response_model=schemas.AgentChat)
async def chat(
app_id: uuid.UUID,
user_id: uuid.UUID,
session_id: uuid.UUID,
query: str,
db=db,
query: schemas.AgentQuery,
auth=Depends(auth),
):
return await agent.chat(app_id=app_id, user_id=user_id, query=query, db=db)
print(query)
return await agent.chat(
app_id=app_id, user_id=user_id, session_id=session_id, query=query
)
@router.get(
@router.post(
"/{session_id}/chat/stream",
responses={
200: {
@ -213,16 +209,20 @@ async def get_chat(
},
)
async def get_chat_stream(
request: Request,
app_id: uuid.UUID,
user_id: uuid.UUID,
session_id: uuid.UUID,
query: str,
db=db,
query: schemas.AgentQuery,
auth=Depends(auth),
):
async def parse_stream():
stream = await agent.stream(app_id=app_id, user_id=user_id, query=query, db=db)
stream = await agent.chat(
app_id=app_id,
user_id=user_id,
session_id=session_id,
query=query,
stream=True,
)
async for chunk in stream:
yield chunk.content

5
src/schemas.py
View File

@ -231,5 +231,10 @@ class Document(DocumentBase):
)
class AgentQuery(BaseModel):
queries: str | list[str]
collections: str | list[str] = "honcho"
class AgentChat(BaseModel):
content: str

4
tests/conftest.py
View File

@ -125,12 +125,12 @@ def client(db_session):
async def sample_data(db_session):
"""Helper function to create test data"""
# Create test app
test_app = models.App(name=str(uuid.uuid4()), metadata={})
test_app = models.App(name=str(uuid.uuid4()))
db_session.add(test_app)
await db_session.flush()
# Create test user
test_user = models.User(name=str(uuid.uuid4()), app_id=test_app.id, metadata={})
test_user = models.User(name=str(uuid.uuid4()), app_id=test_app.id)
db_session.add(test_user)
await db_session.flush()

10
tests/routes/test_messages.py
View File

@ -7,7 +7,7 @@ from src import models # Import your SQLAlchemy models
async def test_create_message(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.id, metadata={})
test_session = models.Session(user_id=test_user.id)
db_session.add(test_session)
await db_session.commit()
@ -31,11 +31,11 @@ async def test_create_message(client, db_session, sample_data):
async def test_get_messages(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session and message
test_session = models.Session(user_id=test_user.id, metadata={})
test_session = models.Session(user_id=test_user.id)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.id, content="Test message", is_user=True, metadata={}
session_id=test_session.id, content="Test message", is_user=True
)
db_session.add(test_message)
await db_session.commit()
@ -56,11 +56,11 @@ async def test_get_messages(client, db_session, sample_data):
async def test_update_message(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session and message
test_session = models.Session(user_id=test_user.id, metadata={})
test_session = models.Session(user_id=test_user.id)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.id, content="Test message", is_user=True, metadata={}
session_id=test_session.id, content="Test message", is_user=True
)
db_session.add(test_message)
await db_session.commit()

12
tests/routes/test_metamessages.py
View File

@ -7,11 +7,11 @@ from src import models # Import your SQLAlchemy models
async def test_create_metamessage(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.id, metadata={})
test_session = models.Session(user_id=test_user.id)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.id, content="Test message", is_user=True, metadata={}
session_id=test_session.id, content="Test message", is_user=True
)
db_session.add(test_message)
await db_session.commit()
@ -37,11 +37,11 @@ async def test_create_metamessage(client, db_session, sample_data):
async def test_get_metamessage(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.id, metadata={})
test_session = models.Session(user_id=test_user.id)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.id, content="Test message", is_user=True, metadata={}
session_id=test_session.id, content="Test message", is_user=True
)
db_session.add(test_message)
await db_session.commit()
@ -69,11 +69,11 @@ async def test_get_metamessage(client, db_session, sample_data):
async def test_update_metamessage(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.id, metadata={})
test_session = models.Session(user_id=test_user.id)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.id, content="Test message", is_user=True, metadata={}
session_id=test_session.id, content="Test message", is_user=True
)
db_session.add(test_message)
await db_session.commit()

1880
uv.lock Normal file
View File

File diff suppressed because it is too large Load Diff