electron-rare/conversations
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test prompts management via github

Browse Source
This commit is contained in:
camilleAND
2026-01-08 11:29:17 +01:00
parent 5676ce68c0
commit bfb374664a
9 changed files with 299 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/backend/chat/agents/base.py
+19 -5
View File
@@ -153,9 +153,16 @@ class BaseAgent(Agent):
This class initializes the agent with model from configuration.
"""
def __init__(self, *, model_hrid, **kwargs):
"""Initialize the agent with model configuration from settings."""
_ignored_kwargs = {"model", "system_prompt", "tools", "toolsets"}
def __init__(self, *, model_hrid, use_configuration_system_prompt: bool = True, **kwargs):
"""Initialize the agent with model configuration from settings.
Args:
model_hrid: HRID of the model to use, as defined in LLM_CONFIGURATIONS.
use_configuration_system_prompt: If False, the system prompt defined in the
LLM configuration will be ignored, allowing subclasses to define all
instructions dynamically.
"""
_ignored_kwargs = {"model", "system_prompt", "tools", "toolsets", "instructions"}
if set(kwargs).intersection(_ignored_kwargs):
raise ValueError(f"{_ignored_kwargs} arguments must not be provided.")
@@ -174,7 +181,7 @@ class BaseAgent(Agent):
# and pydantic_ai.models.infer_model()
_model_instance = self.configuration.model_name
_system_prompt = self.configuration.system_prompt
_system_prompt = self.configuration.system_prompt if use_configuration_system_prompt else None
_base_toolset = (
[
FunctionToolset(
@@ -190,4 +197,11 @@ class BaseAgent(Agent):
_tools = [get_pydantic_tools_by_name(tool_name) for tool_name in self.configuration.tools]
super().__init__(model=_model_instance, instructions=_system_prompt, tools=_tools, **kwargs)
agent_kwargs = {
"model": _model_instance,
"tools": _tools,
}
if _system_prompt is not None:
agent_kwargs["instructions"] = _system_prompt
super().__init__(**agent_kwargs, **kwargs)
src/backend/chat/agents/conversation.py
+15 -1
View File
@@ -11,6 +11,7 @@ from pydantic_ai import ModelMessage
from pydantic_ai.models.function import AgentInfo, FunctionModel
from core.enums import get_language_name
from chat.prompts import get_prompt
from .base import BaseAgent
@@ -95,12 +96,25 @@ class ConversationAgent(BaseAgent):
def __init__(self, *, language=None, **kwargs):
"""Initialize the conversation agent."""
super().__init__(**kwargs)
# Ignore the static system_prompt from configuration; rely entirely on
# dynamically registered instructions, including the external prompt.
super().__init__(use_configuration_system_prompt=False, **kwargs)
# Do not call the real model on deployed instances if the setting is enabled
if settings.WARNING_MOCK_CONVERSATION_AGENT:
self._model = FunctionModel(stream_function=mocked_agent_model)
@self.instructions
def assistant_system_prompt() -> str:
"""
Main system prompt for the assistant, loaded from the synced Prompt store.
This uses the `assistant_system` prompt from the external repository if
available, with a fallback to settings.AI_AGENT_INSTRUCTIONS.
"""
return get_prompt("assistant_system")
@self.instructions
def add_the_date() -> str:
"""
src/backend/chat/migrations/0005_prompt.py
+29
View File
@@ -0,0 +1,29 @@
# Generated by Django 5.2.9 on 2026-01-07 10:20
import uuid
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('chat', '0004_chatconversationattachment_and_more'),
]
operations = [
migrations.CreateModel(
name='Prompt',
fields=[
('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
('name', models.CharField(max_length=100, unique=True)),
('version', models.CharField(help_text='Git commit SHA from which this prompt was synced.', max_length=40)),
('content', models.TextField(help_text='Jinja template content of the prompt.')),
('metadata', models.JSONField(blank=True, help_text='Optional JSON metadata loaded from the sidecar .meta.json file.', null=True)),
],
options={
'abstract': False,
},
),
]
src/backend/chat/models.py
+24
View File
@@ -131,3 +131,27 @@ class ChatConversationAttachment(BaseModel):
null=True,
help_text="Original file key if the Markdown from another file",
)
class Prompt(BaseModel):
"""
Model storing LLM prompts synced from an external repository.
Each prompt is identified by a unique `name` (e.g. "assistant_system") and
optionally linked to metadata coming from a sidecar JSON file.
"""
name = models.CharField(max_length=100, unique=True)
version = models.CharField(
max_length=40,
help_text="Git commit SHA from which this prompt was synced.",
)
content = models.TextField(help_text="Jinja template content of the prompt.")
metadata = models.JSONField(
null=True,
blank=True,
help_text="Optional JSON metadata loaded from the sidecar .meta.json file.",
)
def __str__(self) -> str:
return f"{self.name}@{self.version}"
src/backend/chat/prompt_sync.py
+97
View File
@@ -0,0 +1,97 @@
"""Synchronization helpers for external prompt repositories."""
from __future__ import annotations
import json
from typing import Iterable
import requests
from django.conf import settings
from chat.models import Prompt
def _raw_url(path: str, commit_sha: str) -> str:
"""
Build a raw.githubusercontent.com URL for a given file at a specific commit.
"""
base_url = getattr(settings, "PROMPT_BASE_URL", "https://raw.githubusercontent.com")
owner = settings.PROMPT_REPO_OWNER
repo = settings.PROMPT_REPO_NAME
return f"{base_url}/{owner}/{repo}/{commit_sha}/{path}"
def _fetch_text_file(path: str, commit_sha: str) -> str | None:
"""
Fetch a text file from the prompts repository at the given commit.
"""
url = _raw_url(path, commit_sha)
resp = requests.get(url, timeout=5)
if resp.status_code == 200:
return resp.text
return None
def handle_prompt_sync(payload: dict) -> None:
"""
Synchronize prompts from the external repository based on a GitHub push payload.
Strategy V1: full sync for a static list of prompt names.
This is simple and robust, and prompt files are typically small.
"""
commit_sha = payload.get("after")
if not commit_sha:
# Nothing to do if we don't know which commit to sync from
return
# For now, we rely on a static list of prompt names from settings.
# Example: PROMPT_NAMES = ["assistant_system", "summarizer"]
prompt_names: Iterable[str] = getattr(
settings,
"PROMPT_NAMES",
("assistant_system",),
)
for name in prompt_names:
# Try prompts/ subdirectory first, then root as fallback
jinja_paths = [f"prompts/{name}.jinja", f"{name}.jinja"]
meta_paths = [f"prompts/{name}.meta.json", f"{name}.meta.json"]
content = None
for jinja_path in jinja_paths:
content = _fetch_text_file(jinja_path, commit_sha)
if content is not None:
break
if content is None:
# Prompt may have been removed or file missing; we simply skip it for now.
continue
metadata = None
for meta_path in meta_paths:
metadata_text = _fetch_text_file(meta_path, commit_sha)
if metadata_text:
break
else:
metadata_text = None
if metadata_text:
try:
metadata = json.loads(metadata_text)
except json.JSONDecodeError:
# Ignore invalid metadata for robustness
metadata = None
Prompt.objects.update_or_create(
name=name,
defaults={
"version": commit_sha,
"content": content,
"metadata": metadata,
},
)
src/backend/chat/prompts.py
+32
View File
@@ -0,0 +1,32 @@
"""Runtime access helpers for synced prompts."""
from __future__ import annotations
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from chat.models import Prompt
def get_prompt(name: str) -> str:
"""
Return the content of a prompt by name.
Fallback strategy:
- if the prompt is not found in DB, return the legacy setting-based prompt
for known names (e.g. assistant_system).
"""
try:
prompt = Prompt.objects.get(name=name)
return prompt.content
except ObjectDoesNotExist:
# Fallbacks to keep backward compatibility and avoid hard failures.
if name == "assistant_system":
return settings.AI_AGENT_INSTRUCTIONS
if name == "summarizer":
return settings.SUMMARIZATION_SYSTEM_PROMPT
raise
src/backend/chat/views.py
+40 -1
View File
@@ -6,7 +6,8 @@ from uuid import uuid4
from django.conf import settings
from django.core.files.storage import default_storage
from django.http import Http404, StreamingHttpResponse
from django.http import Http404, HttpResponse, StreamingHttpResponse
from django.views.decorators.csrf import csrf_exempt
import langfuse
import magic
@@ -26,6 +27,7 @@ from core.filters import remove_accents
from activation_codes.permissions import IsActivatedUser
from chat import models, serializers
from chat.clients.pydantic_ai import AIAgentService
from chat.prompt_sync import handle_prompt_sync
from chat.serializers import ChatConversationRequestSerializer
logger = logging.getLogger(__name__)
@@ -435,3 +437,40 @@ class ChatConversationAttachmentViewSet(
)
return Response(serializer.data, status=status.HTTP_200_OK)
@csrf_exempt
def github_prompt_webhook(request):
"""
GitHub webhook endpoint to synchronize assistant prompts.
This endpoint expects a GitHub push payload and validates the request
using the X-Hub-Signature-256 header and a shared secret.
"""
signature = request.headers.get("X-Hub-Signature-256")
secret = getattr(settings, "PROMPT_WEBHOOK_SECRET", None)
if not secret or not signature:
return HttpResponse(status=403)
import hashlib
import hmac
import json as _json
mac = hmac.new(
secret.encode(),
msg=request.body,
digestmod=hashlib.sha256,
)
expected = "sha256=" + mac.hexdigest()
if not hmac.compare_digest(expected, signature):
return HttpResponse(status=403)
try:
payload = _json.loads(request.body)
except _json.JSONDecodeError:
return HttpResponse(status=400)
handle_prompt_sync(payload)
return HttpResponse(status=200)
src/backend/conversations/settings.py
+32
View File
@@ -663,6 +663,38 @@ class Base(BraveSettings, Configuration):
environ_prefix=None,
)
# External prompt repository (for agent system prompts, etc.)
PROMPT_REPO_OWNER = values.Value(
None,
environ_name="PROMPT_REPO_OWNER",
environ_prefix=None,
)
PROMPT_REPO_NAME = values.Value(
None,
environ_name="PROMPT_REPO_NAME",
environ_prefix=None,
)
PROMPT_REPO_BRANCH = values.Value(
"main",
environ_name="PROMPT_REPO_BRANCH",
environ_prefix=None,
)
PROMPT_WEBHOOK_SECRET = values.Value(
None,
environ_name="PROMPT_WEBHOOK_SECRET",
environ_prefix=None,
)
PROMPT_BASE_URL = values.Value(
"https://raw.githubusercontent.com",
environ_name="PROMPT_BASE_URL",
environ_prefix=None,
)
PROMPT_NAMES = values.ListValue(
default=["assistant_system"],
environ_name="PROMPT_NAMES",
environ_prefix=None,
)
# Tools
AI_AGENT_TOOLS = values.ListValue(
default=[],
src/backend/core/urls.py
+11 -1
View File
@@ -9,7 +9,12 @@ from rest_framework.routers import DefaultRouter
from core.api import viewsets
from activation_codes import viewsets as activation_viewsets
from chat.views import ChatConversationAttachmentViewSet, ChatViewSet, LLMConfigurationView
from chat.views import (
ChatConversationAttachmentViewSet,
ChatViewSet,
LLMConfigurationView,
github_prompt_webhook,
)
# - Main endpoints
router = DefaultRouter()
@@ -40,4 +45,9 @@ urlpatterns = [
),
),
path(f"api/{settings.API_VERSION}/config/", viewsets.ConfigView.as_view()),
path(
"webhooks/github/prompts",
github_prompt_webhook,
name="github-prompts-webhook",
),
]