Feat: Add Content Engineering

Makefile

@@ -1,4 +1,4 @@
-.PHONY: dev infra backend frontend up down ingest serve
+.PHONY: dev infra backend frontend up down ingest serve lint lint-fix
 
 dev: up
 
@@ -41,3 +41,11 @@ migrate:
 # Run the StackOverflow ingestion pipeline
 ingest:
 	cd backend && PYTHONPATH=$(PWD)/backend ../.venv/bin/python ingestion/stackoverflow_loader.py
+
+# Lint the backend with ruff
+lint:
+	cd backend && ../.venv/bin/ruff check .
+
+# Lint and auto-fix what ruff can fix
+lint-fix:
+	cd backend && ../.venv/bin/ruff check --fix .

backend/agents/graph.py

@@ -1,8 +1,10 @@
-from typing import TypedDict, Annotated, Sequence, Any
-from langchain_core.messages import BaseMessage
 import operator
 import os
-from langgraph.graph import StateGraph, END
+from collections.abc import Sequence
+from typing import Annotated, Any, TypedDict
+
+from langchain_core.messages import BaseMessage
+from langgraph.graph import END, StateGraph
 
 REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379")
 
@@ -61,11 +63,11 @@ def route_query(state: SynapticState) -> str:
 
 
 # Node imports after SynapticState is defined to avoid circular import failure
-from .nodes.triage import triage_node
-from .nodes.rag_agent import rag_agent
-from .nodes.memory_node import memory_node
-from .nodes.orchestrator import orchestrator_node
-from .nodes.writer_node import writer_node
+from .nodes.memory_node import memory_node  # noqa: E402
+from .nodes.orchestrator import orchestrator_node  # noqa: E402
+from .nodes.rag_agent import rag_agent  # noqa: E402
+from .nodes.triage import triage_node  # noqa: E402
+from .nodes.writer_node import writer_node  # noqa: E402
 
 # ── Graph definition ──────────────────────────────────────────────────────────
 

backend/agents/nodes/memory_node.py

@@ -1,7 +1,7 @@
-import asyncio
+from langfuse import observe
+
 from agents.graph import SynapticState
 from memory.manager import MemoryManager
-from langfuse import observe
 
 memory_manager: MemoryManager = None
 

backend/agents/nodes/orchestrator.py

@@ -1,14 +1,17 @@
 import asyncio
-from typing import Any, Optional
-from langchain_core.documents import Document
-from langchain_core.messages import SystemMessage, HumanMessage
-from llm import utility_llm
+from typing import Any
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langfuse import observe, get_client
+
 from agents.graph import SynapticState
+from constants import BEHAVIORAL_GUARDRAILS
+from context.engineer import assemble
+from llm import utility_llm
 from memory.manager import MemoryManager
+
 from . import rag_agent as _rag_agent
-from .rag_agent import format_memory, extract_citations
-from constants import BEHAVIORAL_GUARDRAILS
-from langfuse import observe
+from .rag_agent import extract_citations
 
 ORCHESTRATOR_PROMPT = """
 You are a synthesis assistant. The user's query requires both factual knowledge and conversation history.
@@ -20,18 +23,7 @@
 If conversation history contradicts the documents, prefer the documents but acknowledge the discrepancy.
 """ + BEHAVIORAL_GUARDRAILS
 
-memory_manager: Optional[MemoryManager] = None
-
-
-def _format_docs(docs: list[Document]) -> str:
-    return "\n\n".join(
-        f"[Source: {d.metadata.get('title', 'Unknown')}]\n{d.page_content}"
-        for d in docs
-    )
-
-
-def _format_long_term(long_term: list[dict[str, Any]]) -> str:
-    return "\n\n".join(m.get("summary", "") for m in long_term if m.get("summary"))
+memory_manager: MemoryManager | None = None
 
 
 class Orchestrator:
@@ -70,21 +62,35 @@ async def orchestrator_node(state: SynapticState) -> dict[str, Any]:
     """
     assert memory_manager is not None, "memory_manager not initialised"
 
+    langfuse = get_client()
+
     memory_result, docs = await asyncio.gather(
         memory_manager.load(session_id=state["session_id"], query=state["query"]),
         _rag_agent.retriever.ainvoke(state["query"]),
     )
 
-    short_term = format_memory(memory_result.get("short_term_memory", []))
-    long_term = _format_long_term(memory_result.get("long_term_memory", []))
-    docs_context = _format_docs(docs)
+    bundle = assemble(
+        "orchestrator_node",
+        short_term_memory=memory_result.get("short_term_memory", []),
+        long_term_memory=memory_result.get("long_term_memory", []),
+        retrieved_chunks=docs,
+    )
+
+    langfuse.update_current_span(
+        metadata={
+            "budget_decision": bundle.decision,
+            "token_count": bundle.token_count,
+            "budget_exceeded": state.get("budget_exceeded"),
+        }
+    )
+
     callbacks = state.get("callbacks", [])
 
     answer = await _orchestrator.merge(
         query=state["query"],
-        docs_context=docs_context,
-        short_term=short_term,
-        long_term=long_term,
+        docs_context=bundle.chunks_context,
+        short_term=bundle.short_term_context,
+        long_term=bundle.long_term_context,
         callbacks=callbacks,
     )
 
@@ -93,4 +99,8 @@ async def orchestrator_node(state: SynapticState) -> dict[str, Any]:
         "retrieved_chunks": [{"content": d.page_content, **d.metadata} for d in docs],
         "citations": extract_citations(docs),
         "final_answer": answer,
+        "token_counts": bundle.token_count,
+        "total_tokens": bundle.total_tokens,
+        "budget_exceeded": bundle.budget_exceeded,
+        "decision": bundle.decision,
     }

backend/agents/nodes/rag_agent.py

@@ -1,10 +1,13 @@
-from agents.graph import SynapticState
-from chain.rag_chain import RagChain
+from typing import Any
+
 from langchain_core.documents import Document
 from langchain_huggingface import HuggingFaceEmbeddings
-from typing import Any
 from langfuse import observe
 
+from agents.graph import SynapticState
+from chain.rag_chain import RagChain
+from context.engineer import assemble
+
 rag_chain = None
 retriever = None
 
@@ -17,36 +20,37 @@ def init(embeddings: HuggingFaceEmbeddings) -> None:
 
 
 @observe(name="rag_agent_node")
-async def rag_agent(state: SynapticState):
-    memory_context = format_memory(state["short_term_memory"])
+async def rag_agent(state: SynapticState) -> dict[str, Any]:
+
+    bundle = assemble(
+        "rag_agent",
+        short_term_memory=state["short_term_memory"],
+        long_term_memory=state["long_term_memory"],
+    )
     callbacks = state.get("callbacks", [])
     result = await rag_chain.ainvoke(
-        {"question": state["query"], "memory_context": memory_context},
+        {"question": state["query"], "memory_context": bundle.memory_context},
         config={"callbacks": callbacks},
     )
+
+    token_counts = {
+        **bundle.token_count,
+        "retrieved_chunks": result.get("chunks_tokens", 0),
+    }
+
     return {
         "final_answer": result["answer"],
         "retrieved_chunks": result["source_documents"],
         "citations": extract_citations(result["source_documents"]),
         "condensed_query": result.get("condensed_query", state["query"]),
+        "token_counts": token_counts,
+        "total_tokens": sum(token_counts.values()),
+        "decision": bundle.decision,
+        "budget_exceeded": bundle.budget_exceeded
+        or result.get("chunks_truncated", False),
     }
 
 
-def format_memory(short_term_memory: list[dict[str, str]]) -> str:
-
-    memory = ""
-    for message in short_term_memory:
-        role: str
-
-        if message["role"] == "ai":
-            role = "AI"
-        else:
-            role = "Human"
-        memory += f"{role}: {message['content']}\n"
-
-    return memory
-
-
 def extract_citations(documents: list[Document]) -> list[dict[str, Any]]:
     citations = []
     for doc in documents:

backend/agents/nodes/triage.py

@@ -1,10 +1,12 @@
-from pydantic import BaseModel
-from langchain_core.messages import SystemMessage, HumanMessage
-from typing import Any
-import re
 import logging
-from agents.graph import SynapticState
+import re
+from typing import Any
+
+from langchain_core.messages import HumanMessage, SystemMessage
 from langfuse import observe
+from pydantic import BaseModel
+
+from agents.graph import SynapticState
 from llm import utility_llm
 
 logger = logging.getLogger(__name__)

backend/agents/nodes/writer_node.py

@@ -1,6 +1,7 @@
+from langfuse import observe
+
 from agents.graph import SynapticState
 from memory.manager import MemoryManager
-from langfuse import observe
 
 memory_manager: MemoryManager = None
 

backend/chain/rag_chain.py

@@ -1,14 +1,15 @@
 from typing import Any
 
-from langchain_huggingface import HuggingFaceEmbeddings
-from langchain_core.documents import Document
-from langchain_core.prompts import ChatPromptTemplate
 from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts import ChatPromptTemplate
 from langchain_core.runnables import RunnableLambda
+from langchain_huggingface import HuggingFaceEmbeddings
+
+from constants import SYSTEM_PROMPT
+from context.engineer import AGENT_BUDGET, fit_chunks
 from ingestion.stackoverflow_loader import CONN_STR
 from llm import main_llm, utility_llm
 from retrieval.chunks_retriever import ChunksRetriever
-from constants import SYSTEM_PROMPT
 
 _CONDENSATION_SYSTEM_PROMPT = (
     "Rewrite the follow-up question as a standalone question by replacing any "
@@ -31,12 +32,6 @@ def __init__(self, embeddings: HuggingFaceEmbeddings) -> None:
         )
         self.prompt = self._build_prompt()
 
-    def _format_docs(self, docs: list[Document]) -> str:
-        return "\n\n".join(
-            f"[Source: {doc.metadata.get('title', 'Unknown')}]\n{doc.page_content}"
-            for doc in docs
-        )
-
     def _build_prompt(self) -> ChatPromptTemplate:
         return ChatPromptTemplate.from_messages(
             [
@@ -60,7 +55,7 @@ def _build_condensation_chain(self) -> Any:
         )
         return prompt | utility_llm | StrOutputParser()
 
-    def build(self):
+    def build(self) -> Any:
         _NO_CONTEXT_REPLY = "I couldn't find relevant information for your question."
         _condensation_chain = self._build_condensation_chain()
         _llm_chain = self.prompt | main_llm | StrOutputParser()
@@ -100,17 +95,31 @@ async def _retrieve(inputs: dict) -> dict:
         async def _answer(inputs: dict) -> dict:
             docs = inputs.get("source_documents", [])
             if not docs:
-                return {**inputs, "context": "", "answer": _NO_CONTEXT_REPLY}
+                return {
+                    **inputs,
+                    "context": "",
+                    "answer": _NO_CONTEXT_REPLY,
+                    "chunks_tokens": 0,
+                    "chunks_truncated": False,
+                }
 
-            context = self._format_docs(docs)
+            context, chunks_tokens, chunks_truncated = fit_chunks(
+                docs, AGENT_BUDGET["rag_agent"]["retrieved_chunks"]
+            )
             answer = await _llm_chain.ainvoke(
                 {
                     "question": inputs["question"],
                     "memory_context": inputs["memory_context"],
                     "context": context,
                 }
             )
-            return {**inputs, "answer": answer, "context": context}
+            return {
+                **inputs,
+                "answer": answer,
+                "context": context,
+                "chunks_tokens": chunks_tokens,
+                "chunks_truncated": chunks_truncated,
+            }
 
         return (
             RunnableLambda(_condense)