ryan/simbarag
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

Merge pull request 'Match llama-server speed and context handling for images and multi-turn chat' (#48) from agent-image-and-context-fixes into main

Browse Source 
Reviewed-on: #48
This commit was merged in pull request #48.
This commit is contained in:
ryan
2026-09-12 15:15:24 -04:00
parent a172a71a70 ca37c4a780
commit a591cfa343
7 changed files with 115 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
blueprints/conversation/__init__.py
+35 -18
View File
@@ -10,7 +10,7 @@ from quart_jwt_extended import (
)
import blueprints.users.models
from utils.image_process import analyze_user_image
from utils.image_process import image_content_block
from utils.image_upload import ImageValidationError, process_image
from utils.s3_client import generate_presigned_url as s3_presigned_url
from utils.s3_client import get_image as s3_get_image
@@ -19,12 +19,15 @@ from utils.s3_client import upload_image as s3_upload_image
from .agents import main_agent
from .logic import (
add_message_to_conversation,
persist_agent_tool_messages,
tool_result_text,
)
from .memory import get_memories_for_user
from .models import (
Conversation,
PydConversation,
PydListConversation,
Speaker,
)
from .prompts import SIMBA_SYSTEM_PROMPT
@@ -64,27 +67,29 @@ async def _build_system_prompt_with_memories(user_id: str) -> str:
def _build_messages_payload(
conversation,
query_text: str,
image_description: str | None = None,
image_block: dict | None = None,
system_prompt: str | None = None,
) -> list:
recent_messages = (
conversation.messages[-10:]
if len(conversation.messages) > 10
conversation.messages[-20:]
if len(conversation.messages) > 20
else conversation.messages
)
messages_payload = [{"role": "system", "content": system_prompt or _SYSTEM_PROMPT}]
for msg in recent_messages[:-1]: # Exclude the message we just added
role = "user" if msg.speaker == "user" else "assistant"
role = "user" if msg.speaker == Speaker.USER else "assistant"
text = msg.text
if msg.image_key and role == "user":
text = f"[User sent an image]\n{text}"
messages_payload.append({"role": role, "content": text})
# Build the current user message with optional image description
if image_description:
content = f"[Image analysis: {image_description}]"
if query_text:
content = f"{query_text}\n\n{content}"
# Attach the image directly to the current message so the (multimodal)
# agent model sees it in one call, instead of a separate describe pass.
if image_block:
content = [
{"type": "text", "text": query_text or "The user sent this image."},
image_block,
]
else:
content = query_text
messages_payload.append({"role": "user", "content": content})
@@ -121,6 +126,7 @@ async def query():
response = await main_agent.ainvoke(payload, config=agent_config)
message = response.get("messages", [])[-1].content
await persist_agent_tool_messages(conversation, user, response.get("messages", []))
await add_message_to_conversation(
conversation=conversation,
message=message,
@@ -190,20 +196,19 @@ async def stream_query():
image_key=image_key,
)
# If an image was uploaded, analyze it with the vision model
image_description = None
# If an image was uploaded, attach it to the agent request directly
image_block = None
if image_key:
try:
image_bytes, _ = await s3_get_image(image_key)
image_description = await analyze_user_image(image_bytes)
logging.info(f"Image analysis complete for {image_key}")
image_block = image_content_block(image_bytes)
except Exception as e:
logging.error(f"Failed to analyze image: {e}")
image_description = "[Image could not be analyzed]"
logging.error(f"Failed to load image {image_key}: {e}")
query_text = f"{query_text or ''}\n\n[An image was sent but could not be loaded.]".strip()
system_prompt = await _build_system_prompt_with_memories(str(user.id))
messages_payload = _build_messages_payload(
conversation, query_text or "", image_description, system_prompt=system_prompt
conversation, query_text or "", image_block, system_prompt=system_prompt
)
payload = {"messages": messages_payload}
agent_config = {
@@ -233,6 +238,15 @@ async def stream_query():
elif event_type == "on_tool_end":
yield f"data: {json.dumps({'type': 'tool_end', 'tool': event['name']})}\n\n"
# Persist tool output so later turns keep the fetched context
output = event.get("data", {}).get("output")
if output is not None:
await add_message_to_conversation(
conversation=conversation,
message=tool_result_text(event["name"], output),
speaker="tool",
user=user,
)
elif event_type == "on_chat_model_stream":
chunk = event.get("data", {}).get("chunk")
@@ -298,10 +312,13 @@ async def get_conversation(conversation_id: str):
# Manually serialize the conversation with messages
messages = []
for msg in conversation.messages:
# Tool messages hold raw fetched context for the LLM; only surface
# their header line (e.g. "[web_search result]") in the UI.
text = msg.text.split("\n", 1)[0] if msg.speaker == Speaker.TOOL else msg.text
messages.append(
{
"id": str(msg.id),
"text": msg.text,
"text": text,
"speaker": msg.speaker.value,
"created_at": msg.created_at.isoformat(),
"image_key": msg.image_key,
blueprints/conversation/logic.py
+36
View File
@@ -1,11 +1,47 @@
from datetime import datetime, timezone
import tortoise.exceptions
from langchain_core.messages import ToolMessage
import blueprints.users.models
from .models import Conversation, ConversationMessage
# Cap persisted tool output so replayed history stays within the LLM context
# budget (llama-server slots are 32k tokens; base prompt + tools already use ~6k).
TOOL_RESULT_MAX_CHARS = 4000
def tool_result_text(tool_name: str, output) -> str:
"""Format a tool's output for persistence as a conversation message."""
content = getattr(output, "content", output)
if isinstance(content, (list, tuple)):
content = "\n".join(str(item) for item in content)
text = str(content)
if len(text) > TOOL_RESULT_MAX_CHARS:
text = text[:TOOL_RESULT_MAX_CHARS] + "\n[truncated]"
return f"[{tool_name} result]\n{text}"
async def persist_agent_tool_messages(
conversation: Conversation,
user: blueprints.users.models.User,
response_messages: list,
) -> None:
"""Persist ToolMessages from an agent response as 'tool' conversation messages.
Keeps fetched context (web searches, RAG chunks, etc.) available to later
turns, which otherwise only see the final text answers.
"""
for msg in response_messages:
if isinstance(msg, ToolMessage):
await add_message_to_conversation(
conversation=conversation,
message=tool_result_text(msg.name or "tool", msg),
speaker="tool",
user=user,
)
async def create_conversation(name: str = "") -> Conversation:
conversation = await Conversation.create(name=name)
blueprints/conversation/models.py
+1
View File
@@ -11,6 +11,7 @@ from tortoise.models import Model
class Speaker(enum.Enum):
USER = "user"
SIMBA = "simba"
TOOL = "tool"
class Conversation(Model):
blueprints/email/__init__.py
+7 -2
View File
@@ -13,8 +13,10 @@ from blueprints.users.models import User
from blueprints.conversation.logic import (
get_conversation_for_channel,
add_message_to_conversation,
persist_agent_tool_messages,
)
from blueprints.conversation.agents import main_agent
from blueprints.conversation.models import Speaker
from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
from . import models # noqa: F401 — register Tortoise ORM models
from .helpers import generate_email_token, get_user_email_address # noqa: F401
@@ -193,17 +195,20 @@ async def webhook():
# Build messages payload
try:
messages = await conversation.messages.all()
recent_messages = list(messages)[-10:]
recent_messages = list(messages)[-20:]
messages_payload = [{"role": "system", "content": SIMBA_SYSTEM_PROMPT}]
for msg in recent_messages[:-1]:
role = "user" if msg.speaker == "user" else "assistant"
role = "user" if msg.speaker == Speaker.USER else "assistant"
messages_payload.append({"role": role, "content": msg.text})
messages_payload.append({"role": "user", "content": body})
logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages")
response = await main_agent.ainvoke({"messages": messages_payload})
response_text = response.get("messages", [])[-1].content
await persist_agent_tool_messages(
conversation, user, response.get("messages", [])
)
except Exception as e:
logger.error(f"Error invoking agent for email: {e}")
response_text = "Sorry, I'm having trouble thinking right now."
blueprints/imessage/__init__.py
+18 -14
View File
@@ -12,7 +12,9 @@ from blueprints.users.models import User
from blueprints.conversation.logic import (
get_conversation_for_channel,
add_message_to_conversation,
persist_agent_tool_messages,
)
from blueprints.conversation.models import Speaker
from blueprints.conversation.agents import main_agent
from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
@@ -200,37 +202,36 @@ async def webhook():
user=user,
)
# Analyze image if present
image_description = None
# Attach image directly to the agent request if present
image_block = None
if media_url:
try:
from utils.image_process import analyze_user_image
from utils.image_process import image_content_block
async with httpx.AsyncClient() as client:
resp = await client.get(media_url, timeout=30)
resp.raise_for_status()
image_bytes = resp.content
image_description = await analyze_user_image(image_bytes)
logger.info(f"Image analysis complete for iMessage from {from_number}")
image_block = image_content_block(resp.content)
except Exception as e:
logger.error(f"Failed to analyze iMessage image: {e}")
image_description = "[Image could not be analyzed]"
logger.error(f"Failed to load iMessage image: {e}")
content = f"{content or ''}\n\n[An image was sent but could not be loaded.]".strip()
# Build messages payload for LangChain agent
try:
messages = await conversation.messages.all()
recent_messages = list(messages)[-10:]
recent_messages = list(messages)[-20:]
messages_payload = [{"role": "system", "content": SIMBA_SYSTEM_PROMPT}]
for msg in recent_messages[:-1]:
role = "user" if msg.speaker == "user" else "assistant"
role = "user" if msg.speaker == Speaker.USER else "assistant"
messages_payload.append({"role": role, "content": msg.text})
if image_description:
user_content = f"[Image analysis: {image_description}]"
if content:
user_content = f"{content}\n\n{user_content}"
if image_block:
user_content = [
{"type": "text", "text": content or "The user sent this image."},
image_block,
]
else:
user_content = content
messages_payload.append({"role": "user", "content": user_content})
@@ -238,6 +239,9 @@ async def webhook():
logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages")
response = await main_agent.ainvoke({"messages": messages_payload})
response_text = response.get("messages", [])[-1].content
await persist_agent_tool_messages(
conversation, user, response.get("messages", [])
)
except Exception as e:
logger.error(f"Error invoking agent: {e}")
blueprints/whatsapp/__init__.py
+8 -3
View File
@@ -11,7 +11,9 @@ from blueprints.users.models import User
from blueprints.conversation.logic import (
get_conversation_for_channel,
add_message_to_conversation,
persist_agent_tool_messages,
)
from blueprints.conversation.models import Speaker
from blueprints.conversation.agents import main_agent
from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
@@ -178,16 +180,16 @@ async def webhook():
# Build messages payload for LangChain agent with system prompt and conversation history
try:
# Get last 10 messages for conversation history
# Get last 20 messages for conversation history
messages = await conversation.messages.all()
recent_messages = list(messages)[-10:]
recent_messages = list(messages)[-20:]
# Build messages payload
messages_payload = [{"role": "system", "content": SIMBA_SYSTEM_PROMPT}]
# Add recent conversation history (exclude the message we just added)
for msg in recent_messages[:-1]:
role = "user" if msg.speaker == "user" else "assistant"
role = "user" if msg.speaker == Speaker.USER else "assistant"
messages_payload.append({"role": role, "content": msg.text})
# Add current query
@@ -197,6 +199,9 @@ async def webhook():
logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages")
response = await main_agent.ainvoke({"messages": messages_payload})
response_text = response.get("messages", [])[-1].content
await persist_agent_tool_messages(
conversation, user, response.get("messages", [])
)
# Log YNAB availability
if os.getenv("YNAB_ACCESS_TOKEN"):
utils/image_process.py
+10 -40
View File
@@ -87,50 +87,20 @@ def _to_jpeg_bytes(file_bytes: bytes) -> bytes:
return buf.getvalue()
async def analyze_user_image(file_bytes: bytes) -> str:
"""Analyze an image uploaded by a user and return a text description.
def image_content_block(file_bytes: bytes) -> dict:
"""Build an OpenAI-style image_url content part from raw image bytes.
Uses llama-server (OpenAI-compatible API) with vision support.
Falls back to OpenAI if llama-server is not configured.
Normalizes to JPEG so HEIC/HEIF and other formats work, then encodes as a
base64 data URL suitable for a multimodal chat message.
"""
import base64
from openai import AsyncOpenAI
file_bytes = _to_jpeg_bytes(file_bytes)
llama_url = os.getenv("LLAMA_SERVER_URL")
if llama_url:
aclient = AsyncOpenAI(base_url=llama_url, api_key="not-needed")
model = os.getenv("LLAMA_MODEL_NAME", "llama-3.1-8b-instruct")
else:
aclient = AsyncOpenAI()
model = "gpt-4o-mini"
b64 = base64.b64encode(file_bytes).decode("utf-8")
response = await aclient.chat.completions.create(
model=model,
messages=[
{
"role": "system",
"content": "You are a helpful image analyst. Describe what you see in the image in detail. Be thorough but concise.",
},
{
"role": "user",
"content": [
{"type": "text", "text": "Please describe this image in detail."},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{b64}",
},
},
],
},
],
)
return response.choices[0].message.content
jpeg_bytes = _to_jpeg_bytes(file_bytes)
b64 = base64.b64encode(jpeg_bytes).decode("utf-8")
return {
"type": "image_url",
"image_url": {"url": f"data:image/jpeg;base64,{b64}"},
}
if __name__ == "__main__":