fix: address all Qodo code review issues in Agno integration

Package & distribution
- pyproject.toml: add integrations* to packages.find include so pip
  install semantica[agno] ships the integration

context_store.py
- upsert_memory(): run NERExtractor after store() to index entities
  into the ContextGraph
- delete_memory() / drop_table() / clear(): call AgentContext.forget()
  to propagate deletions to vector/graph storage
- find_precedents(): pass limit parameter to find_precedents_advanced()
- retrieve(): pass limit as max_results to AgentContext.retrieve()
- add get_context_for_prompt() for automatic system-prompt injection

knowledge_graph.py
- __init__: wire graph_builder.graph_store = self._graph so build()
  persists into the ContextGraph
- add internal AgentContext for vector retrieval (shared ContextGraph)
- search(): use AgentContext.retrieve() for vector similarity; keyword
  scoring as fallback
- _ingest_text(): add paragraph-level chunking before NER/relation
  extraction (parse → split → NER → relation extract → graph build)
- get_graph_context(): return structured subgraph with edge types via
  ContextGraph.get_neighbors()
- load_urls(): validate scheme (http/https only) to prevent SSRF

decision_kit.py
- check_policy(): replace broken PolicyEngine.check_compliance() call
  with inline _eval_rule() that evaluates simple field-op-value rules;
  return compliant=False (not True) on failure — closes security bug

kg_toolkit.py
- add_to_graph(): fix add_node(node_id=, node_type=) and
  add_edge(source_id=, target_id=, edge_type=) to match real API
- query_graph(): use find_nodes() (no label param) + keyword filter
- find_related(): use get_neighbors(node_id=) returning List[Dict]
- infer_facts() / export_subgraph(): use find_nodes() public API
  instead of private _nodes dict

shared_context.py
- _AgentScopedStore: store shared context as self._context (not
  self._ctx) so all inherited AgnoContextStore methods work correctly

tests/integrations/agno/test_kg_toolkit.py
- _FakeGraph: rewrite to match real ContextGraph signatures —
  find_nodes(node_type=), add_node(node_id, node_type, **),
  add_edge(source_id, target_id, edge_type, **),
  get_neighbors(node_id, hops=1, ...) returning List[Dict]

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
KaifAhmad1
2026-03-18 04:21:48 +05:30
co-authored by Claude Sonnet 4.6
parent e315ad849d
commit b2a2d24b14
7 changed files with 396 additions and 111 deletions
+82 -5
View File
@@ -7,10 +7,13 @@ sessions.
Key behaviours
--------------
- ``upsert_memory()`` → stores text in ``AgentContext`` (vector index + graph node)
- ``read_memories()`` → hybrid retrieval: vector similarity + graph hop expansion
- ``upsert_memory()`` → stores text in ``AgentContext`` (vector + graph) and
extracts entities into the knowledge graph
- ``read_memories()`` → hybrid retrieval: vector similarity + graph expansion
- ``delete_memory()`` → removes from cache and calls ``AgentContext.forget()``
- ``record_decision()`` → records a structured decision with reasoning & outcome
- ``find_precedents()`` → returns semantically similar historical decisions
- ``get_context_for_prompt()`` → formats precedents for system-prompt injection
Install
-------
@@ -200,8 +203,9 @@ class AgnoContextStore(_MemoryDbBase): # type: ignore[misc]
"""
Persist ``memory`` into both the vector store and the context graph.
If ``decision_tracking`` is enabled a lightweight decision entry is
also recorded so the memory participates in precedent search.
Entity extraction is performed so the knowledge graph is populated
with nodes for the stored content. If ``decision_tracking`` is enabled
a lightweight decision entry is also recorded.
"""
mem_id = getattr(memory, "id", None) or str(uuid.uuid4())
mem_text = getattr(memory, "memory", str(memory))
@@ -216,6 +220,23 @@ class AgnoContextStore(_MemoryDbBase): # type: ignore[misc]
except Exception as exc: # pragma: no cover
logger.warning("AgentContext.store() failed: %s", exc)
# Extract entities and index them into the knowledge graph
try:
from semantica.semantic_extract import NERExtractor
ner = NERExtractor()
entities = ner.extract_entities(mem_text) or []
kg = getattr(self._context, "knowledge_graph", None)
if kg is not None:
for ent in entities:
name = getattr(ent, "name", str(ent))
ntype = getattr(ent, "type", "Entity")
try:
kg.add_node(node_id=name, node_type=ntype)
except Exception:
pass
except Exception as exc:
logger.debug("NER/graph indexing skipped: %s", exc)
# Optional decision tracking
if self.decision_tracking:
try:
@@ -238,14 +259,26 @@ class AgnoContextStore(_MemoryDbBase): # type: ignore[misc]
def delete_memory(self, id: str) -> None:
self._memories.pop(id, None)
try:
self._context.forget(memory_id=id)
except Exception as exc:
logger.debug("forget(%s) failed: %s", id, exc)
logger.debug("delete_memory id=%s", id)
def drop_table(self) -> None:
self._memories.clear()
try:
self._context.forget()
except Exception as exc:
logger.debug("drop_table forget() failed: %s", exc)
logger.debug("AgnoContextStore: all memories dropped")
def clear(self) -> bool:
self._memories.clear()
try:
self._context.forget()
except Exception as exc:
logger.debug("clear forget() failed: %s", exc)
return True
# ------------------------------------------------------------------
@@ -282,6 +315,7 @@ class AgnoContextStore(_MemoryDbBase): # type: ignore[misc]
return self._context.find_precedents_advanced(
scenario=scenario,
category=category,
limit=limit,
)
except Exception as exc:
logger.warning("find_precedents failed: %s", exc)
@@ -290,11 +324,54 @@ class AgnoContextStore(_MemoryDbBase): # type: ignore[misc]
def retrieve(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
"""Hybrid retrieval: vector similarity + optional graph expansion."""
try:
return self._context.retrieve(query)
return self._context.retrieve(query, max_results=limit)
except Exception as exc:
logger.warning("retrieve failed: %s", exc)
return []
def get_context_for_prompt(self, scenario: str, max_precedents: int = 3) -> str:
"""
Return formatted precedents suitable for injection into a system prompt.
Call this before each LLM invocation to surface relevant past decisions
automatically.
Parameters
----------
scenario:
Description of the current situation.
max_precedents:
Maximum number of precedents to include.
Returns
-------
str
Multi-line string ready to prepend to a system prompt, or an
empty string when no relevant precedents exist.
"""
try:
precedents = self.find_precedents(scenario, limit=max_precedents)
if not precedents:
return ""
lines = ["Relevant past decisions:"]
for i, p in enumerate(precedents[:max_precedents], 1):
if isinstance(p, dict):
sc = p.get("scenario", "")
outcome = p.get("outcome", "")
conf = p.get("confidence", "")
else:
sc = getattr(p, "scenario", str(p))
outcome = getattr(p, "outcome", "")
conf = getattr(p, "confidence", "")
lines.append(
f"{i}. Scenario: {sc} → Outcome: {outcome}"
+ (f" (confidence: {conf})" if conf != "" else "")
)
return "\n".join(lines)
except Exception as exc:
logger.warning("get_context_for_prompt failed: %s", exc)
return ""
@property
def context(self) -> Any:
"""Direct access to the underlying ``AgentContext``."""
+57 -16
View File
@@ -33,6 +33,7 @@ get_decision_summary — Summarise decision history by category
from __future__ import annotations
import json
import re
from typing import Any, Dict, List, Optional
from semantica.utils.logging import get_logger
@@ -308,14 +309,21 @@ class AgnoDecisionKit(_ToolkitBase): # type: ignore[misc]
"""
Validate a proposed decision against policy rules.
Rules are evaluated inline using simple comparison expressions. This
avoids misuse of ``PolicyEngine.check_compliance`` (which requires a
stored ``Decision`` + ``policy_id``) and ensures exceptions never
silently return ``compliant=True``.
Parameters
----------
decision_data:
JSON string describing the decision (must include ``category``,
``outcome``, ``confidence`` keys at minimum).
policy_rules:
JSON list of policy rule strings, e.g.
JSON list of rule strings, e.g.
``'["confidence >= 0.7", "category != \\"test\\""]'``.
Each rule is a simple comparison: ``<field> <op> <value>``
where op is one of ``>=``, ``<=``, ``!=``, ``==``, ``>``, ``<``.
Returns
-------
@@ -325,7 +333,13 @@ class AgnoDecisionKit(_ToolkitBase): # type: ignore[misc]
try:
data = json.loads(decision_data) if isinstance(decision_data, str) else decision_data
except json.JSONDecodeError as exc:
return json.dumps({"error": f"Invalid decision_data JSON: {exc}"})
return json.dumps(
{
"compliant": False,
"violations": [f"Invalid decision_data JSON: {exc}"],
"warnings": [],
}
)
rules: List[str] = []
if policy_rules:
@@ -334,21 +348,48 @@ class AgnoDecisionKit(_ToolkitBase): # type: ignore[misc]
except json.JSONDecodeError:
rules = [r.strip() for r in policy_rules.split(",") if r.strip()]
try:
from semantica.context import PolicyEngine # lazy import
violations: List[str] = []
warnings: List[str] = []
engine = PolicyEngine(graph_store=self._ctx.knowledge_graph) # type: ignore[attr-defined]
result = engine.check_compliance(data, rules)
return json.dumps(
{
"compliant": getattr(result, "compliant", True),
"violations": getattr(result, "violations", []),
"warnings": getattr(result, "warnings", []),
}
)
except Exception as exc:
logger.warning("check_policy failed: %s", exc)
return json.dumps({"compliant": True, "violations": [], "warnings": [], "note": str(exc)})
for rule in rules:
try:
if not self._eval_rule(rule, data):
violations.append(f"Rule violated: {rule}")
except Exception as exc:
warnings.append(f"Could not evaluate rule '{rule}': {exc}")
compliant = len(violations) == 0
logger.debug("check_policy: compliant=%s, violations=%d", compliant, len(violations))
return json.dumps(
{
"compliant": compliant,
"violations": violations,
"warnings": warnings,
}
)
def _eval_rule(self, rule: str, data: Dict[str, Any]) -> bool:
"""Evaluate a simple comparison rule (``field op value``) against data."""
m = re.match(r"(\w+)\s*(>=|<=|!=|==|>|<)\s*(.+)", rule.strip())
if not m:
return True # unrecognised format — pass through
field, op, val_str = m.group(1), m.group(2), m.group(3).strip().strip("\"'")
actual = data.get(field)
if actual is None:
return True # field absent — cannot evaluate
try:
val: Any = type(actual)(val_str)
except (ValueError, TypeError):
val = val_str
ops = {
">=": lambda a, b: a >= b,
"<=": lambda a, b: a <= b,
"!=": lambda a, b: a != b,
"==": lambda a, b: a == b,
">": lambda a, b: a > b,
"<": lambda a, b: a < b,
}
return ops[op](actual, val)
def get_decision_summary(
self,
+60 -33
View File
@@ -20,7 +20,7 @@ Tools exposed
extract_entities — Extract named entities from text
extract_relations — Extract relationships between entities
add_to_graph — Add entities / relations to the context graph
query_graph — Query the graph (natural-language or Cypher)
query_graph — Query the graph (natural-language keyword or Cypher)
find_related — Find concepts related to a given entity
infer_facts — Apply rules to infer new facts from the graph
export_subgraph — Export a subgraph as JSON-LD / RDF Turtle
@@ -235,7 +235,8 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
name = ent.get("name", str(ent))
ntype = ent.get("type", "Entity")
try:
self._graph.add_node(label=name, node_type=ntype) # type: ignore[attr-defined]
# ContextGraph.add_node(node_id, node_type, content=None, **props)
self._graph.add_node(node_id=name, node_type=ntype) # type: ignore[attr-defined]
nodes_added += 1
except Exception:
pass
@@ -248,9 +249,10 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
for rel in rel_list:
src = rel.get("source", "")
tgt = rel.get("target", "")
rel_type = rel.get("relation", "RELATED_TO")
rel_type = rel.get("relation", "related_to")
try:
self._graph.add_edge(src, tgt, edge_type=rel_type) # type: ignore[attr-defined]
# ContextGraph.add_edge(source_id, target_id, edge_type, **props)
self._graph.add_edge(source_id=src, target_id=tgt, edge_type=rel_type) # type: ignore[attr-defined]
edges_added += 1
except Exception:
pass
@@ -264,9 +266,10 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
"""
Query the context graph in natural language or Cypher.
For natural-language queries a keyword-based node lookup is performed.
Pass a string starting with ``"MATCH"`` for raw Cypher execution
(requires a Neo4j / FalkorDB backend).
For natural-language queries all nodes are retrieved and filtered by
whether ``query`` appears in their ``node_id``. Pass a string starting
with ``"MATCH"`` for raw Cypher execution (requires a Neo4j / FalkorDB
backend).
Parameters
----------
@@ -286,18 +289,26 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
records = result if isinstance(result, list) else [str(result)]
return json.dumps({"results": records, "query_type": "cypher"})
except AttributeError:
return json.dumps({"error": "Cypher queries require a Neo4j/FalkorDB backend", "query_type": "cypher"})
return json.dumps(
{
"error": "Cypher queries require a Neo4j/FalkorDB backend",
"query_type": "cypher",
}
)
else:
# Natural-language keyword lookup
nodes = self._graph.find_nodes(label=query) # type: ignore[attr-defined]
out = [
{
"label": getattr(n, "label", str(n)),
"type": getattr(n, "node_type", ""),
"id": getattr(n, "id", ""),
}
for n in (nodes or [])
]
# Natural-language keyword lookup — ContextGraph.find_nodes() → List[Dict]
all_nodes = self._graph.find_nodes() # type: ignore[attr-defined]
q_lower = query.lower()
out = []
for n in (all_nodes or []):
if isinstance(n, dict):
node_id = n.get("node_id", "")
node_type = n.get("node_type", "")
else:
node_id = getattr(n, "id", getattr(n, "label", str(n)))
node_type = getattr(n, "node_type", "")
if q_lower in node_id.lower() or q_lower in node_type.lower():
out.append({"label": node_id, "type": node_type, "id": node_id})
return json.dumps({"results": out, "count": len(out), "query_type": "keyword"})
except Exception as exc:
logger.warning("query_graph failed: %s", exc)
@@ -328,10 +339,14 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
next_frontier: List[str] = []
for e in frontier:
try:
neighbours = self._graph.get_neighbours(e) # type: ignore[attr-defined]
# ContextGraph.get_neighbors(node_id, hops=1, ...) → List[Dict]
neighbours = self._graph.get_neighbors(node_id=e, hops=1) # type: ignore[attr-defined]
for n in (neighbours or []):
label = getattr(n, "label", str(n))
if label not in visited:
if isinstance(n, dict):
label = n.get("node_id", "")
else:
label = getattr(n, "label", str(n))
if label and label not in visited:
visited.add(label)
next_frontier.append(label)
related.append(label)
@@ -376,13 +391,18 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
fact_list = [f.strip() for f in facts.split(",") if f.strip()]
if not fact_list:
# Derive facts from graph nodes
# Derive facts from graph nodes via the public API
try:
nodes = getattr(self._graph, "_nodes", {})
for nid, node in list(nodes.items())[:50]:
label = getattr(node, "label", str(nid))
ntype = getattr(node, "node_type", "Entity")
fact_list.append(f"{ntype}({label})")
all_nodes = self._graph.find_nodes() # type: ignore[attr-defined]
for node in (all_nodes or [])[:50]:
if isinstance(node, dict):
label = node.get("node_id", "")
ntype = node.get("node_type", "Entity")
else:
label = getattr(node, "label", str(node))
ntype = getattr(node, "node_type", "Entity")
if label:
fact_list.append(f"{ntype}({label})")
except Exception:
pass
@@ -422,17 +442,24 @@ class AgnoKGToolkit(_ToolkitBase): # type: ignore[misc]
from semantica.export import RDFExporter # lazy import
exporter = RDFExporter()
rdf_format = {"ttl": "turtle", "json-ld": "json-ld", "xml": "xml", "nt": "nt"}.get(format, format)
rdf_format = {"ttl": "turtle", "json-ld": "json-ld", "xml": "xml", "nt": "nt"}.get(
format, format
)
output = exporter.export_to_rdf(self._graph, format=rdf_format) # type: ignore[arg-type]
return json.dumps({"format": rdf_format, "data": output})
except Exception as exc:
logger.warning("export_subgraph failed: %s", exc)
# Fallback: return graph as plain JSON
# Fallback: return graph nodes via the public API
try:
nodes = [
{"id": getattr(n, "id", k), "label": getattr(n, "label", k)}
for k, n in getattr(self._graph, "_nodes", {}).items()
]
all_nodes = self._graph.find_nodes() # type: ignore[attr-defined]
nodes = []
for n in (all_nodes or []):
if isinstance(n, dict):
nodes.append({"id": n.get("node_id", ""), "label": n.get("node_id", "")})
else:
nodes.append(
{"id": getattr(n, "id", ""), "label": getattr(n, "label", "")}
)
return json.dumps({"format": "json", "nodes": nodes, "note": str(exc)})
except Exception:
return json.dumps({"format": format, "data": "", "error": str(exc)})
+167 -36
View File
@@ -113,6 +113,8 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
Connection URI for the chosen graph store backend.
num_documents:
Default number of documents returned by ``search()``.
chunk_size:
Maximum characters per text chunk during ingestion.
"""
def __init__(
@@ -124,25 +126,39 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
graph_store_backend: str = "inmemory",
graph_store_uri: Optional[str] = None,
num_documents: int = 5,
chunk_size: int = 1000,
**kwargs: Any,
) -> None:
if AGNO_AVAILABLE:
super().__init__(**kwargs) # type: ignore[call-arg]
self.num_documents = num_documents
self.chunk_size = chunk_size
self._graph_store_backend = graph_store_backend
# Lazy imports to keep semantica core optional at import time
from semantica.context import ContextGraph
from semantica.context import AgentContext, ContextGraph
from semantica.kg import GraphBuilder
from semantica.semantic_extract import NERExtractor, RelationExtractor
from semantica.vector_store import VectorStore
self._graph = context_graph or ContextGraph()
# Connect GraphBuilder to the ContextGraph so build() persists content.
self._graph_builder = graph_builder or GraphBuilder()
self._graph_builder.graph_store = self._graph
self._ner = ner_extractor or NERExtractor()
self._rel = relation_extractor or RelationExtractor()
# In-process document store for search fallback
# Internal AgentContext for vector-based retrieval (shares same graph).
self._agent_context = AgentContext(
vector_store=VectorStore(backend="faiss"),
knowledge_graph=self._graph,
decision_tracking=False,
)
# In-process document store for keyword-search fallback
self._docs: List[Dict[str, Any]] = []
logger.info(
@@ -163,14 +179,39 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
"""
Multi-hop GraphRAG search.
1. Vector retrieval over stored document texts.
1. Vector retrieval via ``AgentContext.retrieve()``.
2. Graph hop expansion for entities found in top results.
3. Returns a list of Agno ``Document`` objects.
Falls back to keyword scoring over the in-process ``_docs`` cache
when vector retrieval is unavailable.
"""
k = num_documents or self.num_documents
results: List[Any] = []
# Simple keyword / substring filter over in-process store
# Primary: vector similarity retrieval
try:
retrieved = self._agent_context.retrieve(query, max_results=k)
for item in retrieved:
if isinstance(item, dict):
content = item.get("content", item.get("text", str(item)))
entities = item.get("entities", [])
meta = {k2: v for k2, v in item.items() if k2 not in ("content", "text")}
else:
content = str(item)
entities = []
meta = {}
extra = self._graph_context_for(entities) if entities else ""
if extra:
content = content + "\n\n[Graph context]\n" + extra
results.append(AgnoDocument(content=content, meta_data=meta))
if results:
logger.debug("search('%s') → %d documents (vector)", query, len(results))
return results
except Exception as exc:
logger.debug("Vector retrieval failed, using keyword fallback: %s", exc)
# Fallback: keyword / substring scoring over in-process cache
q_lower = query.lower()
scored = [
(doc, sum(1 for w in q_lower.split() if w in doc["text"].lower()))
@@ -180,12 +221,10 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
top = [d for d, _ in scored[:k]]
for doc in top:
# Graph expansion: pull related entities from the context graph
extra = self._graph_context_for(doc.get("entities", []))
content = doc["text"]
if extra:
content += "\n\n[Graph context]\n" + extra
results.append(
AgnoDocument(
content=content,
@@ -195,7 +234,7 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
)
)
logger.debug("search('%s') → %d documents", query, len(results))
logger.debug("search('%s') → %d documents (keyword)", query, len(results))
return results
def load(
@@ -236,10 +275,22 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
self.load_urls(urls)
def load_urls(self, urls: List[str]) -> None:
"""Fetch each URL and ingest the response body."""
"""Fetch each URL and ingest the response body.
Only ``http`` and ``https`` schemes are permitted to prevent SSRF.
"""
import urllib.request
from urllib.parse import urlparse
for url in urls:
parsed = urlparse(url)
if parsed.scheme not in ("http", "https"):
logger.warning(
"Skipping URL with disallowed scheme '%s': %s",
parsed.scheme,
url,
)
continue
try:
with urllib.request.urlopen(url, timeout=10) as resp: # noqa: S310
text = resp.read().decode("utf-8", errors="replace")
@@ -261,52 +312,125 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
self._ingest_text(text, source=source)
def get_graph_context(self, entity: str) -> str:
"""Return a text summary of an entity's subgraph (neighbours + edges)."""
return self._graph_context_for([entity])
"""
Return a structured text representation of an entity's subgraph
(neighbours and edge types), suitable for structured reasoning.
Parameters
----------
entity:
Root entity name (must have been added to the graph).
Returns
-------
str
Multi-line text with nodes and labelled edge types.
"""
lines = [f"Entity: {entity}"]
try:
neighbours = self._graph.get_neighbors(node_id=entity, hops=1)
for n in (neighbours or [])[:10]:
if isinstance(n, dict):
node_id = n.get("node_id", "")
ntype = n.get("node_type", "")
edge_type = n.get("edge_type", "related_to")
suffix = f" (type: {ntype})" if ntype else ""
lines.append(f" --[{edge_type}]--> {node_id}{suffix}")
else:
lines.append(f" --> {getattr(n, 'label', str(n))}")
except Exception:
pass
return "\n".join(lines)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _chunk_text(self, text: str) -> List[str]:
"""Split text into chunks at paragraph boundaries."""
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
if not paragraphs:
return [text] if text.strip() else []
chunks: List[str] = []
current: List[str] = []
current_len = 0
for para in paragraphs:
if current_len + len(para) > self.chunk_size and current:
chunks.append("\n\n".join(current))
current = []
current_len = 0
current.append(para)
current_len += len(para)
if current:
chunks.append("\n\n".join(current))
return chunks or [text]
def _ingest_text(self, text: str, source: str = "<text>") -> None:
"""Run the full extraction pipeline and store in graph + doc list."""
import uuid
# NER
entities: List[str] = []
chunks = self._chunk_text(text)
all_entities: List[str] = []
all_relations: List[Any] = []
for chunk in chunks:
# NER
ner_result: List[Any] = []
try:
ner_result = self._ner.extract_entities(chunk) or []
chunk_entities = [getattr(e, "name", str(e)) for e in ner_result]
all_entities.extend(chunk_entities)
except Exception as exc:
logger.debug("NER failed for chunk in '%s': %s", source, exc)
# Relation extraction
try:
chunk_relations = self._rel.extract_relations(chunk, entities=ner_result) or []
all_relations.extend(chunk_relations)
except Exception as exc:
logger.debug("RelationExtractor failed for chunk in '%s': %s", source, exc)
# Graph build — graph_store is wired to self._graph in __init__
try:
ner_result = self._ner.extract_entities(text)
entities = [
getattr(e, "name", str(e)) for e in (ner_result or [])
sources = [
{
"text": text,
"entities": all_entities,
"relations": all_relations,
"source": source,
}
]
except Exception as exc:
logger.debug("NER failed for '%s': %s", source, exc)
# Relation extraction
relations: List[Any] = []
try:
relations = self._rel.extract_relations(text, entities=ner_result) # type: ignore[arg-type]
except Exception as exc:
logger.debug("RelationExtractor failed for '%s': %s", source, exc)
# Graph build
try:
sources = [{"text": text, "entities": entities, "relations": relations, "source": source}]
self._graph_builder.build(sources)
except Exception as exc:
logger.debug("GraphBuilder.build() failed for '%s': %s", source, exc)
# Cache document for search
# Vector index for AgentContext.retrieve()
try:
self._agent_context.store(text, conversation_id=source)
except Exception as exc:
logger.debug("AgentContext.store() failed for '%s': %s", source, exc)
# Cache document for keyword-search fallback
self._docs.append(
{
"id": str(uuid.uuid4()),
"text": text,
"source": source,
"entities": entities,
"entities": all_entities,
"metadata": {"source": source},
}
)
logger.debug("Ingested '%s'%d entities, %d relations", source, len(entities), len(relations))
logger.debug(
"Ingested '%s'%d entities, %d relations, %d chunks",
source,
len(all_entities),
len(all_relations),
len(chunks),
)
def _ingest_path(self, path: Path, recursive: bool = False) -> None:
"""Walk a file or directory and ingest all text files."""
@@ -334,11 +458,18 @@ class AgnoKnowledgeGraph(_KnowledgeBase): # type: ignore[misc]
lines: List[str] = []
for entity in entities[:3]: # limit to avoid context bloat
try:
nodes = self._graph.find_nodes(label=entity) # type: ignore[attr-defined]
for node in (nodes or [])[:3]:
label = getattr(node, "label", entity)
ntype = getattr(node, "node_type", "")
lines.append(f"- {label} ({ntype})" if ntype else f"- {label}")
neighbours = self._graph.get_neighbors(node_id=entity, hops=1)
for n in (neighbours or [])[:3]:
if isinstance(n, dict):
node_id = n.get("node_id", "")
ntype = n.get("node_type", "")
edge_type = n.get("edge_type", "related_to")
lines.append(
f"- {entity} --[{edge_type}]--> {node_id}"
+ (f" ({ntype})" if ntype else "")
)
else:
lines.append(f"- {entity} --> {getattr(n, 'label', str(n))}")
except Exception:
pass
return "\n".join(lines)
+8 -4
View File
@@ -58,14 +58,17 @@ class _AgentScopedStore(AgnoContextStore):
def __init__(self, shared: "AgnoSharedContext", role: str) -> None:
# Re-use the parent's context rather than creating a new one.
# We skip the normal __init__ and wire directly.
# We skip the normal __init__ and wire all required parent attributes
# directly so that inherited methods (record_decision, find_precedents,
# retrieve, get_context_for_prompt) work correctly via self._context.
self._role = role
self._shared = shared
self._memories: Dict[str, Any] = {}
self.decision_tracking = shared.decision_tracking
self.graph_expansion = shared.graph_expansion
self.session_id = f"{shared.session_id}::{role}"
self._ctx = shared._context # shared AgentContext
# Use the attribute name the parent class expects.
self._context = shared._context # type: ignore[attr-defined]
# ------------------------------------------------------------------
# Override upsert / record to tag with role
@@ -78,13 +81,13 @@ class _AgentScopedStore(AgnoContextStore):
mem_text = getattr(memory, "memory", str(memory))
try:
self._ctx.store(mem_text, conversation_id=self.session_id)
self._context.store(mem_text, conversation_id=self.session_id)
except Exception as exc:
logger.warning("[%s] store failed: %s", self._role, exc)
if self.decision_tracking:
try:
self._ctx.record_decision(
self._context.record_decision(
category=f"memory:{self._role}",
scenario=mem_text[:200],
reasoning=f"Stored by agent role='{self._role}'",
@@ -258,6 +261,7 @@ class AgnoSharedContext:
return self._context.find_precedents_advanced(
scenario=scenario,
category=category,
limit=limit,
)
except Exception as exc:
logger.warning("find_precedents failed: %s", exc)
+1 -1
View File
@@ -213,7 +213,7 @@ semantica-worker = "semantica.worker:main"
# ---------------- TOOLING ----------------
[tool.setuptools.packages.find]
where = ["."]
include = ["semantica*"]
include = ["semantica*", "integrations*"]
[tool.black]
line-length = 88
+21 -16
View File
@@ -85,27 +85,32 @@ class _FakeReasoner:
class _FakeGraph:
"""Fake ContextGraph whose signatures match the real ContextGraph API."""
def __init__(self):
self._nodes = {}
self._edges = []
self._node_store: dict = {} # node_id -> {"node_id": ..., "node_type": ...}
self._edge_store: list = []
def find_nodes(self, label=None):
node = MagicMock()
node.label = label or "SomeNode"
node.node_type = "Entity"
node.id = "n1"
return [node]
# ContextGraph.find_nodes(node_type=None) -> List[Dict]
def find_nodes(self, node_type=None):
nodes = list(self._node_store.values())
if node_type:
nodes = [n for n in nodes if n.get("node_type") == node_type]
return nodes
def add_node(self, label, node_type="Entity"):
self._nodes[label] = MagicMock(label=label, node_type=node_type)
# ContextGraph.add_node(node_id, node_type, content=None, **props) -> bool
def add_node(self, node_id, node_type="Entity", content=None, **props):
self._node_store[node_id] = {"node_id": node_id, "node_type": node_type}
return True
def add_edge(self, src, tgt, edge_type="RELATED_TO"):
self._edges.append((src, tgt, edge_type))
# ContextGraph.add_edge(source_id, target_id, edge_type, **props) -> bool
def add_edge(self, source_id, target_id, edge_type="related_to", **props):
self._edge_store.append((source_id, target_id, edge_type))
return True
def get_neighbours(self, entity):
n = MagicMock()
n.label = f"Neighbour_of_{entity}"
return [n]
# ContextGraph.get_neighbors(node_id, hops=1, ...) -> List[Dict]
def get_neighbors(self, node_id, hops=1, relationship_types=None, min_weight=0.0):
return [{"node_id": f"Neighbour_of_{node_id}", "node_type": "Entity"}]
class TestAgnoKGToolkitInit(unittest.TestCase):