Implementation-level technical reference
NDC KB Agent Code Patterns
KB Agent Code Patterns
Building knowledge-augmented agents sounds straightforward until you try it at scale. You bolt a vector database onto a language model, throw in some retrieval logic, and watch it hallucinate confidently about documents it never found.
The patterns in this document come from building production knowledge agents at Acme CyberSecurity - agents that handle thousands of threat assessment queries daily, route between specialized knowledge domains, and need to be right because security decisions depend on them.
These aren't theoretical patterns. They're the solutions that emerged after the obvious approaches failed.
The Challenge: Knowledge Agents at Scale
A basic RAG (Retrieval Augmented Generation) system works fine for simple Q&A. User asks a question, you retrieve relevant documents, you generate an answer. Done.
But real knowledge systems face harder problems:
Domain complexity. Users ask questions that span multiple knowledge domains. A query about endpoint protection might need information from threat intelligence, product documentation, vendor comparisons, and deployment guides. No single retrieval strategy works for all of these.
Query ambiguity. "Tell me about our firewall options" could mean: compare products, explain technical specs, summarize deployment requirements, or generate a vendor response. The same words need different handling based on intent.
Authority conflicts. Different sources have different reliability. A vendor's published specifications are authoritative for that product. Internal security assessments are authoritative for threat analysis. Blog posts are illustrative but not definitive. The agent needs to know which sources to trust for which questions.
Error compounding. In a complex knowledge system, errors propagate. A bad retrieval leads to a wrong answer, which might inform a follow-up that compounds the error. You need mechanisms to detect and correct mistakes before they cascade.
The patterns below address each of these challenges.
Pattern 1: Dual-Agent Architecture
The first instinct when building a knowledge agent is to create one agent that does everything - retrieves documents, synthesizes answers, handles edge cases. This works until the agent's context window fills up or the task complexity exceeds what a single reasoning chain can handle.
The solution: split into specialized agents with distinct responsibilities.
The Pattern
# src/agents/knowledge_system.py
class SecurityKnowledgeSystem:
"""
Dual-agent architecture: Threat Detection Advisor (TDA) handles
security-specific queries, Security Catalog Expert handles product
and vendor information.
Why two agents? Each maintains domain-specific context and
retrieval strategies. TDA knows threat databases and assessment
frameworks. Catalog Expert knows product specs and vendor comparisons.
"""
def __init__(self, threat_db: ThreatDatabase, catalog_db: CatalogDatabase):
self.tda = ThreatDetectionAdvisor(threat_db)
self.catalog_expert = SecurityCatalogExpert(catalog_db)
self.router = GoalRouter()
def process_query(self, query: str, context: ConversationContext) -> Response:
# First: determine which agent should handle this
goal = self.router.detect_goal(query, context)
# Route to specialized agent
if goal.domain == "threat_assessment":
return self.tda.respond(query, context, goal)
elif goal.domain == "product_catalog":
return self.catalog_expert.respond(query, context, goal)
else:
# Synthesis queries need both agents
return self._synthesize(query, context, goal)
def _synthesize(self, query: str, context: ConversationContext, goal: Goal) -> Response:
"""
Some queries span both domains. Get inputs from each agent,
then synthesize. Example: "What products address the threat
we discussed?" needs threat context + product knowledge.
"""
threat_context = self.tda.get_relevant_context(query, context)
product_options = self.catalog_expert.find_products(threat_context)
return self._generate_synthesis(query, threat_context, product_options)Why This Works
Focused context windows. Each agent maintains context relevant to its domain. The Threat Detection Advisor isn't cluttered with product specifications; the Security Catalog Expert isn't loaded with threat intelligence. This means more relevant context per query.
Domain-specific retrieval. Different knowledge domains benefit from different retrieval strategies. Threat data needs recency weighting - a three-month-old threat report might be outdated. Product catalogs need attribute matching - filtering by capability, certification, or deployment model. Splitting agents lets each optimize for its domain.
Independent evolution. When threat database schemas change or new product categories emerge, you update one agent without touching the other. The system decomposes along natural domain boundaries.
Explicit handoffs. When a query spans domains, the handoff is explicit and traceable. You can see exactly what context passed from one agent to another, which helps debugging.
Implementation Notes
The router is critical - a bad routing decision sends the query to an agent that can't handle it well. More on routing in Pattern 2.
Synthesis queries are the expensive case. They require both agents to contribute, which roughly doubles latency. Design your domain splits to minimize synthesis needs for common queries.
Pattern 2: 4-Question Goal Detection Cascade
Routing queries to the right agent (or the right retrieval strategy) requires understanding user intent. But intent classification is surprisingly hard - the same words can mean different things in different contexts.
The Pattern
The cascade asks four independent questions in sequence. Each question is specific and binary, making classification more reliable than trying to map directly to intent categories.
# src/routing/goal_detector.py
class GoalDetector:
"""
4-Question cascade for goal detection. Each question is independent -
you're not building a decision tree, you're checking specific patterns.
Why a cascade instead of direct classification? Classification accuracy
degrades with more categories. Binary questions are easier to get right.
"""
def detect_goal(self, message: str, context: ConversationContext) -> Goal:
message_lower = message.lower()
# Q1: Does this reference specific threats or vulnerabilities?
# If yes, this is likely a threat assessment query
if self._has_threat_reference(message_lower, context):
if self._needs_product_mapping(message_lower):
return Goal(domain="synthesis", type="threat_to_product", confidence=0.8)
return Goal(domain="threat_assessment", type="lookup", confidence=0.9)
# Q2: Does this have vendor comparison indicators?
# Keywords like "compare", "options", "recommend", "RFP"
if self._has_comparison_indicators(message_lower):
return Goal(domain="product_catalog", type="comparison", confidence=0.85)
# Q3: Does this reference the current conversation context?
# "that product", "the one you mentioned", "those options"
if self._references_prior_context(message_lower, context):
return self._goal_from_context(context)
# Q4: Is this a general security domain question?
# Default path - route based on keyword matching
return self._classify_by_keywords(message_lower)
def _has_threat_reference(self, message: str, context: ConversationContext) -> bool:
"""
Check for explicit threat references: CVE numbers, threat names,
vulnerability descriptions, or references to prior threat discussion.
"""
threat_patterns = [
r'cve-\d{4}-\d+', # CVE references
r'(ransomware|malware|phishing|apt|zero-day)', # Threat categories
r'(vulnerability|exploit|attack vector)', # Security terminology
]
for pattern in threat_patterns:
if re.search(pattern, message):
return True
# Also check if conversation has threat context
return context.has_active_threat_discussion()
def _has_comparison_indicators(self, message: str) -> bool:
"""
Vendor comparison language signals product catalog queries.
"""
indicators = [
'compare', 'comparison', 'versus', 'vs',
'options', 'alternatives', 'recommend',
'rfp', 'vendor', 'pricing', 'features',
'which one', 'best for', 'should we use'
]
return any(ind in message for ind in indicators)Why This Works
Ordered specificity. The questions go from most specific (threat references are unambiguous) to least specific (keyword matching is fuzzy). Specific matches exit early with high confidence.
Independent testing. Each question doesn't depend on previous answers. This means errors don't compound - a wrong answer to Q1 doesn't poison Q2.
Confidence scoring. Each branch returns a confidence level. Downstream logic can handle low-confidence routing differently - maybe requesting clarification or trying multiple agents.
Conversation context. Q3 recognizes that many queries only make sense in context. "Tell me more about that" has no inherent domain, but the prior conversation tells you exactly what "that" refers to.
Implementation Notes
The keyword lists need maintenance as usage patterns evolve. Track queries that route incorrectly and update the pattern recognition.
Confidence thresholds matter. A query that matches Q1 with 0.6 confidence might need different handling than one with 0.95 confidence. Design for uncertainty, not just classification.
Pattern 3: Authority Hierarchy
Not all sources are equally trustworthy, and trust depends on what question you're asking. A vendor's product documentation is authoritative for that product's specifications. An internal security assessment is authoritative for your organization's threat exposure. Published research is authoritative for general vulnerability information.
The Pattern
Define explicit authority levels and query sources accordingly.
# src/retrieval/authority_hierarchy.py
class AuthorityHierarchy:
"""
Authority levels for knowledge sources. Higher authority sources are
queried first and weighted more heavily in synthesis.
Level 1: Authoritative sources (official documentation, specifications)
Level 2: Expert sources (internal assessments, validated research)
Level 3: Informative sources (blog posts, community discussions)
Level 4: Background (general context, may be outdated)
"""
AUTHORITY_CONFIG = {
"threat_assessment": [
{"source": "threat_intelligence_feed", "level": 1, "weight": 1.0},
{"source": "internal_security_assessments", "level": 2, "weight": 0.85},
{"source": "vendor_advisories", "level": 2, "weight": 0.8},
{"source": "security_research", "level": 3, "weight": 0.6},
{"source": "community_reports", "level": 4, "weight": 0.3},
],
"product_catalog": [
{"source": "vendor_documentation", "level": 1, "weight": 1.0},
{"source": "certification_databases", "level": 1, "weight": 0.95},
{"source": "internal_assessments", "level": 2, "weight": 0.8},
{"source": "analyst_reports", "level": 3, "weight": 0.6},
{"source": "user_reviews", "level": 4, "weight": 0.3},
]
}
def retrieve_with_authority(
self,
query: str,
domain: str,
min_level: int = 3
) -> List[Document]:
"""
Retrieve documents respecting authority hierarchy.
Higher-authority sources are queried first; lower sources fill gaps.
"""
sources = self.AUTHORITY_CONFIG.get(domain, [])
results = []
for source_config in sources:
if source_config["level"] > min_level:
continue # Skip low-authority sources unless needed
docs = self._retrieve_from_source(
query,
source_config["source"],
limit=5
)
# Weight documents by source authority
for doc in docs:
doc.authority_weight = source_config["weight"]
results.extend(docs)
# If we have enough authoritative content, stop
if self._sufficient_coverage(results, source_config["level"]):
break
return self._deduplicate_and_rank(results)Why This Works
Explicit trust model. Instead of treating all retrieved documents equally, the system knows which sources to trust for which questions. This reduces hallucination from low-quality sources.
Efficiency. High-authority sources are queried first. If they provide sufficient coverage, lower-authority sources aren't even checked. This reduces latency for common queries.
Tunable behavior. Different use cases can set different minimum authority levels. A quick internal query might accept Level 3 sources. A response going to a client might require Level 1 only.
Traceable answers. Every claim in the response can be traced to a source with a known authority level. This supports auditing and error analysis.
Implementation Notes
Authority levels are domain-specific. A source that's Level 1 for product specifications might be Level 3 for threat assessment. The configuration should reflect this.
The "sufficient coverage" check is nuanced. It's not just about document count - you need to assess whether the retrieved content actually addresses the query. Topic modeling or embedding similarity can help here.
Pattern 4: 3-Branch Correction Flow
Knowledge agents will make mistakes. The question is how to detect and correct them before they reach the user or propagate to downstream systems.
The Pattern
Three types of corrections, each with different handling.
# src/correction/correction_flow.py
class CorrectionFlow:
"""
3-Branch correction for knowledge agent errors:
Branch 1: Retrieval correction - wrong documents retrieved
Branch 2: Synthesis correction - right documents, wrong answer
Branch 3: Authority correction - answer conflicts with higher authority
Each branch has different detection signals and remediation steps.
"""
def check_and_correct(
self,
query: str,
retrieved_docs: List[Document],
generated_response: str,
context: ConversationContext
) -> CorrectionResult:
# Branch 1: Did we retrieve the right documents?
retrieval_check = self._check_retrieval_quality(query, retrieved_docs)
if not retrieval_check.passed:
return self._correct_retrieval(query, retrieval_check.issues)
# Branch 2: Does the response actually answer the query?
synthesis_check = self._check_synthesis_quality(
query, retrieved_docs, generated_response
)
if not synthesis_check.passed:
return self._correct_synthesis(query, retrieved_docs, synthesis_check.issues)
# Branch 3: Does the response conflict with authoritative sources?
authority_check = self._check_authority_conflicts(
generated_response,
retrieved_docs
)
if not authority_check.passed:
return self._correct_authority(query, authority_check.conflicts)
return CorrectionResult(corrected=False, response=generated_response)
def _check_retrieval_quality(
self,
query: str,
docs: List[Document]
) -> CheckResult:
"""
Retrieval quality checks:
- Do retrieved docs have semantic similarity to query?
- Do they cover the query's key concepts?
- Are there obvious missing topics?
"""
query_embedding = self.embedder.embed(query)
key_concepts = self._extract_key_concepts(query)
issues = []
# Check semantic similarity
for doc in docs:
similarity = cosine_similarity(query_embedding, doc.embedding)
if similarity < 0.3: # Threshold tuned empirically
issues.append(f"Low similarity doc: {doc.id} ({similarity:.2f})")
# Check concept coverage
covered_concepts = self._concepts_in_docs(docs)
missing = key_concepts - covered_concepts
if missing:
issues.append(f"Missing concepts: {missing}")
return CheckResult(passed=len(issues) == 0, issues=issues)
def _correct_retrieval(self, query: str, issues: List[str]) -> CorrectionResult:
"""
Retrieval correction: expand search, try different strategies,
or acknowledge gaps.
"""
# Try query expansion
expanded_query = self._expand_query(query)
new_docs = self.retriever.retrieve(expanded_query, limit=10)
# Retry with new docs
if self._check_retrieval_quality(query, new_docs).passed:
return CorrectionResult(
corrected=True,
correction_type="retrieval_expansion",
new_docs=new_docs
)
# If still failing, acknowledge the gap
return CorrectionResult(
corrected=True,
correction_type="acknowledged_gap",
response=self._generate_gap_response(query, issues)
)Why This Works
Ordered checks. Retrieval errors are caught before synthesis runs on bad data. Authority conflicts are caught after synthesis, when there's a specific claim to check. The order matters.
Specific remediation. Each branch has its own fix. Retrieval problems need query expansion or different sources. Synthesis problems need regeneration with better prompting. Authority conflicts need explicit correction with citations.
Graceful degradation. When correction fails, the system acknowledges the gap rather than confidently presenting a wrong answer. "I couldn't find authoritative information on this" is better than hallucination.
Debugging support. Corrections are logged with type and reason. When analyzing errors, you can see which branch triggered and why, which guides system improvements.
Implementation Notes
The thresholds (like 0.3 similarity) are empirical. Start conservative and tune based on real query logs.
Authority checking requires extractable statements. If the response is too vague to check ("Security is important"), authority correction can't help. This creates pressure toward specific, checkable responses - which is good.
Pattern 5: Cache-Stable Context Assembly
Knowledge agents often need context beyond the immediate query - conversation history, user profile, document excerpts. Assembling this context is expensive. Caching helps, but context changes frequently. The solution is cache-stable assembly that minimizes invalidation.
The Pattern
# src/context/stable_assembly.py
class CacheStableContextAssembler:
"""
Assemble context from multiple sources while maximizing cache hits.
Key insight: decompose context into stable and volatile components.
Stable components (document excerpts, user profile) cache well.
Volatile components (recent messages, current query) don't.
Assembly combines cached stable parts with fresh volatile parts.
"""
def __init__(self, cache: ContextCache):
self.cache = cache
self.ttl_config = {
"document_excerpt": 3600, # 1 hour - documents rarely change
"user_profile": 300, # 5 min - profile updates are rare
"conversation_summary": 60, # 1 min - summarize, not store raw
"query_context": 0, # Never cache - always fresh
}
def assemble_context(
self,
query: str,
user_id: str,
conversation: Conversation,
retrieved_docs: List[Document]
) -> AssembledContext:
"""
Build context from stable (cached) and volatile (fresh) components.
"""
context_parts = []
cache_stats = {"hits": 0, "misses": 0}
# Stable: document excerpts (cache per document ID)
for doc in retrieved_docs:
excerpt = self._get_or_create_excerpt(doc, cache_stats)
context_parts.append(("document", excerpt))
# Stable: user profile (cache per user ID)
profile = self._get_or_create_profile(user_id, cache_stats)
if profile:
context_parts.append(("user_profile", profile))
# Semi-stable: conversation summary (short TTL)
conv_summary = self._get_or_create_summary(conversation, cache_stats)
context_parts.append(("conversation", conv_summary))
# Volatile: current query (never cached)
context_parts.append(("query", self._format_query(query)))
return AssembledContext(
parts=context_parts,
cache_hit_rate=cache_stats["hits"] / (cache_stats["hits"] + cache_stats["misses"])
)
def _get_or_create_excerpt(
self,
doc: Document,
stats: Dict
) -> str:
"""
Document excerpts are computed once and cached by document ID.
Excerpts include the most relevant sections, not the full document.
"""
cache_key = f"excerpt:{doc.id}:{doc.version}"
cached = self.cache.get(cache_key)
if cached:
stats["hits"] += 1
return cached
stats["misses"] += 1
excerpt = self._compute_excerpt(doc)
self.cache.set(cache_key, excerpt, ttl=self.ttl_config["document_excerpt"])
return excerptWhy This Works
Cache granularity. Instead of caching entire contexts (which invalidate whenever anything changes), cache components independently. Document excerpts cache regardless of conversation state.
Version-aware keys. Cache keys include version identifiers. When a document updates, the old cache entry simply expires - no explicit invalidation needed.
TTL tuning. Different components have different change frequencies. Documents are stable (long TTL). Conversations change constantly (short TTL or no cache). The TTL configuration makes this explicit.
Hit rate visibility. The assembler reports cache hit rate. This lets you monitor efficiency and tune TTL values based on actual usage patterns.
Implementation Notes
Excerpt computation is the expensive part. The caching pays off when the same documents are retrieved repeatedly - common in knowledge systems where certain core documents are highly relevant.
Consider pre-warming caches for frequently-accessed documents. If you know certain threat intelligence feeds are queried constantly, compute and cache their excerpts proactively.
Anti-Patterns to Avoid
After building systems with these patterns, certain anti-patterns become obvious.
Single-agent monolith. One agent that handles everything becomes impossible to debug and optimize. The context window fills with irrelevant information. Domain-specific retrieval strategies can't be applied.
Direct intent classification. Trying to classify queries into many categories directly fails at scale. The cascade pattern (binary questions in sequence) is more robust.
Equal source weighting. Treating all retrieved documents equally leads to hallucination from low-quality sources. The authority hierarchy prevents this.
Silent failures. Knowledge agents that confidently present wrong answers are dangerous. The correction flow ensures errors are either corrected or explicitly acknowledged.
Full context caching. Caching assembled contexts seems efficient but leads to constant invalidation. Cache components, not compositions.
Ignoring conversation state. Queries often reference prior conversation. Ignoring this context leads to misrouted queries and irrelevant responses.
Putting It Together
These patterns compose into a complete knowledge agent architecture:
-
Query arrives - Goal detection cascade determines intent and routes to appropriate agent(s)
-
Agent retrieves - Authority hierarchy ensures high-quality sources are prioritized
-
Context assembles - Cache-stable assembly builds context efficiently from stable and volatile components
-
Response generates - Specialized agent produces response using domain-specific knowledge
-
Correction checks - 3-branch flow catches retrieval, synthesis, and authority errors before response is returned
-
Response delivered - User receives answer with appropriate confidence indicators
The patterns aren't independent - they reinforce each other. Authority hierarchy makes correction flow more effective (clear reference points for conflict detection). Goal detection enables efficient caching (routes are stable, so cache keys are predictable). Dual-agent architecture makes each pattern simpler (smaller scope per agent).
Start with the pattern that addresses your biggest pain point. For most systems, that's authority hierarchy - it has the largest impact on response quality with relatively simple implementation.
Then add patterns incrementally. Each one makes the system more robust, but each also adds complexity. Find the balance that works for your scale and requirements.
Related
For patterns on building the harness that wraps these agents, see Claude Code Harness Builder.
For the broader context on skill development, see Building Production-Ready AI Tools with Claude Skills.