mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-08-29 04:26:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feat/crewai integration (#988)

Browse Source 
* feat(crewai): add first-class CrewAI integration (#962)

Add native CrewAI support so Crew agents can share a ContextGraph and
AgentContext via BaseTool subclasses and a BaseKnowledgeSource, matching
the existing agno integration pattern.

- SemanticaKGTool: 5 KG actions (extract_entities, extract_relations,
  add_to_graph, query_graph, find_related) with sync run()/async arun()
- SemanticaDecisionTool: 5 decision-intelligence actions
  (record_decision, find_precedents, trace_causal_chain,
  analyze_impact, check_policy) over AgentContext
- SemanticaKnowledgeSource: serializes a ContextGraph into crew
  knowledge storage; bridges legacy load_content() and current
  validate_content()/aadd() contracts for crewai>=0.80.0
- All classes degrade gracefully when crewai is absent
- New pip extra crewai=... included in the all bundle
- 70 new tests (stub-based present-case + subprocess degradation path)
- Docs: integrations/crewai.md, docs.json nav, README matrix updates

* fix(crewai): harden tools against real Semantica dataclass shapes (#962)

Bugs found during live testing with crewai 1.15.16:

- SemanticaKGTool.add_to_graph crashed on real Entity/Relation dataclasses
  ('str' object has no attribute 'end_char'): string names were passed to
  extract_relations(entities=...), which requires Entity objects, and the
  tool read .name/.source/.target instead of Entity's .text/.label and
  Relation's .subject/.object. Add shape-agnostic field helpers.
- SemanticaDecisionTool() created an AgentContext without a knowledge_graph,
  so _decision_backend was never set and record_decision raised 'Decision
  tracking is not enabled'. Wire in a ContextGraph.
- record_decision hard-failed when the agent omitted optional fields; fall
  back to category='general', reasoning='agent decision',
  outcome='recorded'.

Add tests covering real Entity/Relation dataclass shapes and the live
auto-created AgentContext path (now 77 crewai tests, 212 total).

* fix(crewai): make find_related traverse edges undirected (#962)

ContextGraph.get_neighbors only follows outgoing edges, so a node whose
only edge is incoming (A -> B) reported no related concepts. Rebuild a
bidirectional adjacency from find_edges() in SemanticaKGTool._find_related
so 'related' honors both directions.

* fix(crewai): harden tools for checkpoint serialization and correct action semantics (#962)

- Exclude live graph/context/extractor state from JSON serialization
  (model_dump(mode="json")) so CrewAI checkpointing no longer raises
  PydanticSerializationError; model_post_init self-heals defaults on restore
- query_graph now searches node content via graph.query() plus id/type
- trace_causal_chain returns an explicit error when causal tracing is
  unavailable instead of substituting similarity precedents; call
  trace_decision_causality(..., max_depth=...) with the correct kwarg name
- find_precedents propagates max_precedents/limit to the backend instead of
  being silently capped at 10
- Serialize add_to_graph batches under a module lock to prevent concurrent
  double-counting; skip nameless entities instead of creating repr()-junk nodes
- aadd() runs CPU-bound serialization in a thread executor
- Mirror crewai args_schema serialize/restore in the conftest stub and add
  serialization regression tests (crewai: 92 tests)

* fix(crewai): correct check_policy coercion, guard causal tracing, and harden concurrency (#962)

- _eval_rule now coerces rule values type-aware: bool("false") was truthy, so
  'enabled == false' reported a violation for enabled=false, and string datums
  like "0.90" were compared lexicographically instead of numerically
- _trace_causal_chain no longer raises AttributeError (which escaped _run) when
  the decision context lacks knowledge_graph; returns honest error JSON
- SemanticaKnowledgeSource storage failures log an actionable ERROR; without a
  configured crew embedder agents previously retrieved nothing silently
- add_to_graph uses a per-graph re-entrant lock (WeakKeyDictionary) instead of a
  process-global one: independent graphs no longer serialize each other and
  re-entrant extractor callbacks cannot deadlock
- entity/relation confidence=None normalizes to 1.0 instead of failing the
  whole extraction with float(None)
- add subprocess integration test against real crewai covering Crew-level
  serialization round-trip and checkpoint restore (stub tests cannot see it)
- docs: embedder requirement for SemanticaKnowledgeSource; resume contract note

* fix(crewai): surface knowledge-source save failures at ERROR when storage is wired (#962)

Re-verification against real crewai showed the embedder-missing failure raises
ValueError even though storage IS wired, so the old except-ValueError branch
mislabeled it as 'storage not wired' and logged DEBUG — hiding the failure.
Distinguish by storage presence instead of exception type: storage is None ->
DEBUG keep-in-memory (legitimate standalone use); storage wired but save()
raises -> actionable ERROR. Add regression test mirroring real crewai's
ValueError-on-missing-embedder behavior.

* fix(crewai): expose run()/arun() entry points in degraded mode (#962)

The public crewai contract is run()/arun(); without crewai installed they were
missing (only the private _run existed), so the documented 'usable without
crewai' path raised AttributeError at the entry point. Define them in degraded
mode only, leaving crewai's BaseTool implementations untouched when present.
Extend the degradation subprocess test to exercise run() and arun().

* fix(crewai): standardize query shape, field-name rules, and restore-state flag

- _query_graph: id/type matches now return the same schema as content
  matches (id/type/label/content/score) instead of a bare list
- _eval_rule: non-greedy field capture so hyphen/dot/space JSON keys
  (e.g. "risk-score >= 0.9") are addressable in policy rules
- add had_live_state/reconstructed_state so checkpoint-restored tools
  and knowledge sources signal that their live graph/context was lost
  and an empty one reconstructed; knowledge source no longer hides the
  loss by eagerly rebuilding its graph inside __init__ (pydantic calls
  __init__ during model_validate)

* fix(crewai): address Qodo review — confidence errors, string trim, holistic availability

- record_decision: stop calling float() in _run, so malformed confidence
  values surface as JSON errors (via _record_decision's handling) instead
  of crashing the tool
- _coerce_value: return the stripped string for non-numeric literals so
  whitespace-padded decision_data fields match policy rules
- centralize crewai availability in _availability.py so the exported
  CREWAI_AVAILABLE flag is holistic across tools and knowledge source
  (previously each module probed crewai independently and the package
  flag came from decision_tool only)

* ci: regenerate requirements-ci.txt for the crewai extra

The crewai extra in pyproject.toml brings in crewai, crewai-tools and
transitive deps (chromadb, lancedb, ...). Recompile with
uv==0.12.1 per CONTRIBUTING.md so the CI staleness check passes.

* ci: keep crewai out of the locked CI dependency set

crewai (all versions) hard-requires chromadb~=1.1.0, which carries a
pre-authentication code-injection advisory (CVE-2026-45829 / GHSA-f4j7-r4q5-qw2c)
with NO fixed release — even the latest 1.5.9 is affected. Keeping crewai in
the 'all' extra failed pip-audit and the safety check on requirements-ci.txt.

- drop crewai from the 'all' aggregate (standalone semantica[crewai] extra is
  unchanged and still installs crewai)
- stop listing crewai-tools in the extra: the integration only uses crewai core
  (BaseTool, BaseKnowledgeSource) and crewai-tools pulled extra transitive deps
- regenerate requirements-ci.txt: OSV/pip-audit 0 vulnerabilities, safety 0
  vulnerabilities, staleness check matches

* docs(crewai): document crewai extra scope and chromadb CVE-2026-45829

- CHANGELOG: extra is crewai>=0.80.0 only (no crewai-tools) and is not
  part of the 'all' bundle, with the chromadb CVE-2026-45829 reason
- integrations/crewai/README.md: add a security warning that installing
  the extra pulls chromadb~=1.1.0, which is affected by the unpatched
  pre-auth code-injection CVE-2026-45829

---------
This commit is contained in:
Shinde vinayak rao patil
2026-08-16 11:15:43 +05:00
committed by GitHub
parent 5579851208
commit d94d8f6ab8
18 changed files with 3434 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
integrations/crewai/README.md
+108
View File
@@ -0,0 +1,108 @@
# Semantica × CrewAI
First-class integration between Semantica and [CrewAI](https://github.com/crewAIInc/crewAI) — give your crews a shared semantic knowledge graph, decision intelligence, and graph-based retrieval.
## Installation
```bash
pip install semantica[crewai]
```
Requires `crewai >= 0.80.0`. If `crewai` is not installed, the integration still imports (classes degrade gracefully), but you can't pass the objects to a `Crew`.
> **⚠️ Security note:** crewai hard-requires `chromadb~=1.1.0`, which is currently affected by the unpatched pre-authentication code-injection advisory **CVE-2026-45829** (no fixed release — even the latest chromadb 1.5.9 is affected). Installing `semantica[crewai]` pulls that dependency into your environment. The `crewai` extra is intentionally **not** part of `semantica[all]` for this reason — only install it where you actually use CrewAI, and follow chromadb for a patched release.
## 1. SemanticaKGTool
A `BaseTool` that lets agents **build and query** a shared `ContextGraph` mid-reasoning:
- `extract_entities` — extract named entities from `text`
- `extract_relations` — extract relationships from `text`
- `add_to_graph` — extract entities/relations from `text` and add them to the shared graph
- `query_graph` — keyword-search the graph using `query`
- `find_related` — find concepts related to `entity` within `hops`
```python
from crewai import Agent, Crew, Task
from semantica.context import ContextGraph
from integrations.crewai import SemanticaKGTool
graph = ContextGraph()
analyst = Agent(
role="Knowledge Analyst",
goal="Build and explore a knowledge graph from documents",
backstory="You map entities and relationships into a shared graph.",
tools=[SemanticaKGTool(graph=graph)],
)
crew = Crew(
agents=[analyst],
tasks=[Task(description="Extract and link key entities from the brief", expected_output="JSON", agent=analyst)],
)
result = crew.kickoff()
```
All actions return JSON, so agents get parseable results.
## 2. SemanticaDecisionTool
A `BaseTool` that wraps `AgentContext` and exposes decision intelligence:
- `record_decision` — record a decision with reasoning and outcome
- `find_precedents` — retrieve past decisions similar to a scenario
- `trace_causal_chain` — trace the causal chain from a decision
- `analyze_impact` — assess downstream influence using graph centrality
- `check_policy` — validate a proposed decision against rule-based policies
```python
from crewai import Agent, Crew, Task
from integrations.crewai import SemanticaDecisionTool
planner = Agent(
role="Decision Planner",
goal="Make grounded, precedented decisions",
backstory="You record decisions and validate them against policy.",
tools=[SemanticaDecisionTool()],
)
crew = Crew(agents=[planner], tasks=[...])
```
When no `AgentContext` is passed, one is created in-memory with `decision_tracking=True`.
## 3. SemanticaKnowledgeSource
A `BaseKnowledgeSource` that serializes the current state of a `ContextGraph` (nodes, edges, metadata) into CrewAI's knowledge storage, giving **every agent in the crew** retrieval access to the graph:
```python
from crewai import Agent, Crew, Task
from semantica.context import ContextGraph
from integrations.crewai import SemanticaKnowledgeSource
graph = ContextGraph()
graph.add_node(node_id="privacy", node_type="policy", content="...")
researcher = Agent(
role="Policy Researcher",
goal="Answer questions from the knowledge base",
backstory="You retrieve from graph knowledge to answer accurately.",
)
crew = Crew(
agents=[researcher],
tasks=[...],
knowledge_sources=[SemanticaKnowledgeSource(graph=graph)],
)
```
> **Embedder required:** storing chunks goes through CrewAI's knowledge pipeline, which needs an embedder. Set `Crew(embedder=...)` (or provide CrewAI's default credentials, e.g. `OPENAI_API_KEY`). Without a working embedder, storage fails, an ERROR is logged, and agents retrieve nothing — the crew still runs with empty knowledge queries.
### Compatibility note
CrewAI's `BaseKnowledgeSource` contract changed between `0.80.x` and current releases (`load_content()` → `validate_content()`/`aadd()`). `SemanticaKnowledgeSource` implements both the legacy and current methods, so it works across `crewai>=0.80.0`.
### Sharing state & checkpoints
- Each tool/source holds whatever `graph`/`context` you pass it. When omitted, a fresh in-memory object is created and a warning is logged — instances that auto-create their own state do **not** share knowledge, so pass the same object to every agent that must share.
- Live state (`ContextGraph`, `AgentContext`, extractors) is excluded from CrewAI's JSON serialization. After restoring from a checkpoint, re-attach the live graph/context to the restored objects.
integrations/crewai/__init__.py
+44
View File
@@ -0,0 +1,44 @@
"""
Semantica × CrewAI Integration
==============================
First-class integration between the Semantica semantic intelligence stack and
the `CrewAI <https://github.com/crewAIInc/crewAI>`_ agentic framework.
Public surface
--------------
SemanticaKGTool — CrewAI ``BaseTool`` exposing KG construction/query actions
SemanticaDecisionTool — CrewAI ``BaseTool`` exposing decision-intelligence actions
SemanticaKnowledgeSource— CrewAI ``BaseKnowledgeSource`` giving crews graph knowledge
Quick start
-----------
pip install semantica[crewai]
>>> from integrations.crewai import (
... SemanticaKGTool,
... SemanticaDecisionTool,
... SemanticaKnowledgeSource,
... )
Compatibility
-------------
Requires ``crewai >= 0.80.0``. All three classes degrade gracefully when
``crewai`` is not installed — they are still importable and carry the full
Semantica API, but cannot be passed to ``Crew`` / ``Agent`` constructors.
"""
from ._availability import CREWAI_AVAILABLE, CREWAI_IMPORT_ERROR
from .decision_tool import SemanticaDecisionTool
from .kg_tool import SemanticaKGTool
from .knowledge_source import SemanticaKnowledgeSource
__all__ = [
"SemanticaKGTool",
"SemanticaDecisionTool",
"SemanticaKnowledgeSource",
"CREWAI_AVAILABLE",
"CREWAI_IMPORT_ERROR",
]
__version__ = "0.1.0"
integrations/crewai/_availability.py
+24
View File
@@ -0,0 +1,24 @@
"""
Shared CrewAI availability probe.
Every integration module needs to know whether the real ``crewai`` package is
installed. Probing once here (instead of once per module) guarantees the
exported ``CREWAI_AVAILABLE`` flag means the *whole* integration is ready — a
caller gating on it will never see tools using CrewAI while a knowledge source
silently degrades (or vice versa).
"""
from typing import Optional
CREWAI_AVAILABLE = False
CREWAI_IMPORT_ERROR: Optional[str] = None
try:
from crewai.knowledge.source.base_knowledge_source import ( # noqa: F401
BaseKnowledgeSource,
)
from crewai.tools import BaseTool # noqa: F401
CREWAI_AVAILABLE = True
except ImportError as exc:
CREWAI_IMPORT_ERROR = str(exc)
integrations/crewai/decision_tool.py
+555
View File
@@ -0,0 +1,555 @@
"""
SemanticaDecisionTool — a CrewAI ``BaseTool`` exposing Semantica's decision
intelligence (``AgentContext``) to agents.
Lets agents record decisions with reasoning, retrieve past precedents, trace
causal chains, analyse downstream impact, and validate proposed decisions
against policy rules.
Install
-------
pip install semantica[crewai]
Example
-------
>>> from integrations.crewai import SemanticaDecisionTool
>>> from crewai import Agent, Crew, Task
>>> tool = SemanticaDecisionTool()
>>> crew = Crew(
... agents=[Agent(role="...", goal="...", backstory="...", tools=[tool])],
... tasks=[...],
... )
Tools exposed
-------------
record_decision — Record a decision with reasoning and outcome
find_precedents — Search past decisions similar to a scenario
trace_causal_chain— Trace the causal chain from a decision node
analyze_impact — Assess downstream influence of a decision
check_policy — Validate a proposed decision against policy rules
"""
from __future__ import annotations
import json
import re
from typing import Any, Dict, List, Literal, Optional, Type
from pydantic import BaseModel, Field
from semantica.utils.logging import get_logger
from ._availability import CREWAI_AVAILABLE
logger = get_logger(__name__)
# ---------------------------------------------------------------------------
# Optional: CrewAI BaseTool base class
# ---------------------------------------------------------------------------
_BaseTool: Any = object
if CREWAI_AVAILABLE:
from crewai.tools import BaseTool as _BaseTool # type: ignore
# ---------------------------------------------------------------------------
# Input schema
# ---------------------------------------------------------------------------
class SemanticaDecisionToolInput(BaseModel):
"""
Input schema for ``SemanticaDecisionTool``.
Exactly one action is dispatched per call; the remaining fields are only
used by the actions that need them.
"""
action: Literal[
"record_decision",
"find_precedents",
"trace_causal_chain",
"analyze_impact",
"check_policy",
] = Field(
...,
description=(
"Which decision-intelligence operation to run. One of: "
"'record_decision', 'find_precedents', 'trace_causal_chain', "
"'analyze_impact', 'check_policy'."
),
)
category: Optional[str] = Field(
None,
description="Domain category, e.g. 'loan_approval'. Used by 'record_decision'.",
)
scenario: Optional[str] = Field(
None,
description=(
"Short description of the situation. Used by 'record_decision' and "
"'find_precedents'."
),
)
reasoning: Optional[str] = Field(
None, description="Why this outcome was chosen. Used by 'record_decision'."
)
outcome: Optional[str] = Field(
None, description="The decision result. Used by 'record_decision'."
)
confidence: float = Field(
0.8,
ge=0.0,
le=1.0,
description="Confidence score in [0, 1]. Used by 'record_decision'.",
)
entities: Optional[str] = Field(
None,
description="Comma-separated entity names. Used by 'record_decision'.",
)
decision_id: Optional[str] = Field(
None,
description=(
"Identifier of a decision. Used by 'trace_causal_chain' and "
"'analyze_impact'."
),
)
depth: int = Field(
3,
ge=1,
le=20,
description="Maximum chain depth. Used by 'trace_causal_chain'.",
)
decision_data: Optional[str] = Field(
None,
description=(
"JSON object describing a proposed decision. Used by 'check_policy'."
),
)
policy_rules: Optional[str] = Field(
None,
description=(
"JSON list of rule strings like 'confidence >= 0.7'. Used by "
"'check_policy'."
),
)
# ---------------------------------------------------------------------------
# SemanticaDecisionTool
# ---------------------------------------------------------------------------
class SemanticaDecisionTool(_BaseTool): # type: ignore[misc]
"""
CrewAI tool that surfaces Semantica's decision intelligence as agent actions.
Parameters
----------
context:
A ``semantica.context.AgentContext`` (or compatible object exposing
``record_decision``, ``find_precedents_advanced``,
``analyze_decision_influence``). A fresh in-memory context is created
when ``None``.
max_precedents:
Default number of precedents returned by ``find_precedents``.
causal_depth:
Default chain depth used by ``trace_causal_chain``.
"""
name: str = "semantica_decision"
description: str = (
"Decision intelligence toolkit. Actions: 'record_decision' (record a "
"decision with category, scenario, reasoning, outcome, confidence), "
"'find_precedents' (search past decisions similar to 'scenario'), "
"'trace_causal_chain' (trace the causal chain from 'decision_id'), "
"'analyze_impact' (assess downstream influence of 'decision_id'), "
"'check_policy' (validate 'decision_data' JSON against 'policy_rules' "
"rules like 'confidence >= 0.7'). Returns JSON."
)
args_schema: Type[BaseModel] = SemanticaDecisionToolInput
context: Any = Field(default=None, exclude=True)
max_precedents: int = 5
causal_depth: int = 3
had_live_state: bool = False
reconstructed_state: bool = Field(default=False, exclude=True)
def __init__(
self,
context: Any = None,
max_precedents: int = 5,
causal_depth: int = 3,
**kwargs: Any,
) -> None:
if CREWAI_AVAILABLE:
super().__init__(
context=context,
max_precedents=max_precedents,
causal_depth=causal_depth,
**kwargs,
)
else:
super().__init__()
self.context = context
self.max_precedents = max_precedents
self.causal_depth = causal_depth
# Degraded mode is a plain class — no model_post_init lifecycle.
self._ensure_defaults()
logger.info("SemanticaDecisionTool initialised (crewai=%s)", CREWAI_AVAILABLE)
def model_post_init(self, __context: Any) -> None:
"""Re-create default state after validation/deserialisation.
``context`` is excluded from JSON serialisation (CrewAI checkpoints
serialise every tool via ``model_dump(mode="json")``), so a tool
restored from a checkpoint has ``None`` state until this runs.
"""
self._ensure_defaults()
super().model_post_init(__context)
def _ensure_defaults(self) -> None:
"""Lazy-import and build a real AgentContext when none is wired."""
if self.context is None:
from semantica.context import AgentContext, ContextGraph
from semantica.vector_store import VectorStore
self.context = AgentContext(
vector_store=VectorStore(backend="faiss"),
decision_tracking=True,
knowledge_graph=ContextGraph(),
)
if self.had_live_state:
self.reconstructed_state = True
logger.warning(
"SemanticaDecisionTool: the live decision context was lost "
"during serialization/checkpoint restore — an EMPTY "
"context was reconstructed; re-attach the original context "
"before continuing"
)
else:
logger.warning(
"SemanticaDecisionTool created a fresh in-memory "
"AgentContext — agents sharing decision state must be "
"wired to the same context"
)
self.had_live_state = True
# ------------------------------------------------------------------
# CrewAI entry points
# ------------------------------------------------------------------
def _run(
self,
action: str,
category: Optional[str] = None,
scenario: Optional[str] = None,
reasoning: Optional[str] = None,
outcome: Optional[str] = None,
confidence: float = 0.8,
entities: Optional[str] = None,
decision_id: Optional[str] = None,
depth: int = 3,
decision_data: Optional[str] = None,
policy_rules: Optional[str] = None,
**kwargs: Any,
) -> str:
valid = {
"record_decision",
"find_precedents",
"trace_causal_chain",
"analyze_impact",
"check_policy",
}
if action not in valid:
return json.dumps(
{
"error": f"Unknown action '{action}'. Valid actions: "
+ ", ".join(sorted(valid))
}
)
if action == "record_decision":
return self._record_decision(
category=category or "general",
scenario=scenario or "decision recorded",
reasoning=reasoning or "agent decision",
outcome=outcome or "recorded",
confidence=confidence,
entities=entities,
)
if action == "find_precedents":
return self._find_precedents(scenario=scenario or "", category=category)
if action == "trace_causal_chain":
return self._trace_causal_chain(decision_id or "", depth=depth)
if action == "analyze_impact":
return self._analyze_impact(decision_id or "")
return self._check_policy(decision_data or "", policy_rules)
async def _arun(self, action: str, **kwargs: Any) -> str:
"""Async variant of ``_run`` for CrewAI's async tool path."""
return self._run(action=action, **kwargs)
# ------------------------------------------------------------------
# Actions
# ------------------------------------------------------------------
def _record_decision(
self,
category: str,
scenario: str,
reasoning: str,
outcome: str,
confidence: float = 0.8,
entities: Optional[str] = None,
) -> str:
entity_list: Optional[List[str]] = None
if entities:
entity_list = [e.strip() for e in entities.split(",") if e.strip()]
try:
decision_id = self.context.record_decision(
category=category,
scenario=scenario,
reasoning=reasoning,
outcome=outcome,
confidence=float(confidence),
entities=entity_list,
)
result = {"decision_id": str(decision_id), "status": "recorded"}
logger.info("record_decision → %s", decision_id)
except Exception as exc:
result = {"error": str(exc), "status": "failed"}
logger.warning("record_decision failed: %s", exc)
return json.dumps(result)
def _find_precedents(
self,
scenario: str,
category: Optional[str] = None,
limit: Optional[int] = None,
) -> str:
k = limit if limit is not None else self.max_precedents
try:
precedents = self.context.find_precedents_advanced(
scenario=scenario,
category=category,
limit=k,
)
out: List[Dict[str, Any]] = []
for p in (precedents or [])[:k]:
if isinstance(p, dict):
out.append(p)
else:
out.append(
{
"scenario": getattr(p, "scenario", str(p)),
"outcome": getattr(p, "outcome", ""),
"confidence": getattr(p, "confidence", 0.0),
"category": getattr(p, "category", ""),
}
)
logger.info("find_precedents('%s') → %d results", scenario, len(out))
return json.dumps({"precedents": out, "count": len(out)})
except Exception as exc:
logger.warning("find_precedents failed: %s", exc)
return json.dumps({"precedents": [], "count": 0, "error": str(exc)})
def _trace_causal_chain(self, decision_id: str, depth: Optional[int] = None) -> str:
if not decision_id:
return json.dumps(
{
"error": "decision_id is required for trace_causal_chain",
"causal_chain": [],
"decision_id": "",
}
)
max_depth = depth or self.causal_depth
try:
graph = getattr(self.context, "knowledge_graph", None)
if graph is None:
return json.dumps(
{
"error": (
"causal tracing is not available on this knowledge "
"graph (the decision context has no knowledge_graph)"
),
"causal_chain": [],
"decision_id": decision_id,
}
)
trace = getattr(graph, "trace_decision_causality", None)
if trace is None:
return json.dumps(
{
"error": (
"causal tracing is not available on this knowledge graph "
"(graph.trace_decision_causality is not implemented)"
),
"causal_chain": [],
"decision_id": decision_id,
}
)
chain = trace(decision_id, max_depth=max_depth)
return json.dumps({"causal_chain": chain, "decision_id": decision_id})
except Exception as exc:
logger.warning("trace_causal_chain failed: %s", exc)
return json.dumps(
{"error": str(exc), "causal_chain": [], "decision_id": decision_id}
)
def _analyze_impact(self, decision_id: str) -> str:
try:
influence = self.context.analyze_decision_influence(decision_id)
if not isinstance(influence, dict):
influence = {"influence": str(influence)}
influence["decision_id"] = decision_id
return json.dumps(influence)
except Exception as exc:
logger.warning("analyze_impact failed: %s", exc)
return json.dumps({"error": str(exc), "decision_id": decision_id})
def _check_policy(
self,
decision_data: str,
policy_rules: Optional[str] = None,
) -> str:
try:
data = (
json.loads(decision_data)
if isinstance(decision_data, str)
else decision_data
)
except json.JSONDecodeError as exc:
return json.dumps(
{
"compliant": False,
"violations": [f"Invalid decision_data JSON: {exc}"],
"warnings": [],
}
)
if not isinstance(data, dict):
return json.dumps(
{
"compliant": False,
"violations": [
f"decision_data must decode to a JSON object, "
f"got {type(data).__name__}: {data!r}"
],
"warnings": [],
}
)
violations: List[str] = []
warnings: List[str] = []
rules: List[str] = []
if policy_rules:
try:
parsed_rules = json.loads(policy_rules)
except json.JSONDecodeError:
rules = [r.strip() for r in policy_rules.split(",") if r.strip()]
else:
if isinstance(parsed_rules, str):
rules = [parsed_rules]
elif isinstance(parsed_rules, list):
for item in parsed_rules:
if isinstance(item, str):
rules.append(item)
else:
warnings.append(
f"Ignoring non-string policy rule entry: {item!r}"
)
else:
warnings.append(
f"policy_rules must decode to a JSON list of rule strings, "
f"got {type(parsed_rules).__name__}: {parsed_rules!r}"
)
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.
This is a small standalone evaluator for the tool's ``check_policy``
action — it is intentionally independent of Semantica's policy engine
so agents get a bounded, side-effect-free rule check. Rules are
``<field> <op> <value>`` comparisons only; there is no expression
evaluation (no ``eval``), so untrusted rule strings are safe to pass.
Values are coerced type-aware: ``true``/``false`` (and ``1``/``0``)
become booleans, numeric literals become numbers, and string values
that parse as numbers are compared numerically, so ``score == 0.9``
holds for ``score: "0.90"`` and ``enabled == false`` holds for
``enabled: false``. Field names may contain hyphens, dots and spaces
(e.g. ``risk-score >= 0.9``); they are matched against ``data`` keys
as-is.
"""
m = re.match(r"(.+?)\s*(>=|<=|!=|==|>|<)\s*(.+)$", rule.strip())
if not m:
raise ValueError(f"unrecognised rule format: {rule!r}")
field, op, val_str = m.group(1), m.group(2), m.group(3).strip().strip("\"'")
if field not in data:
raise ValueError(f"rule references undefined field {field!r}")
actual = data[field]
if actual is None:
raise ValueError(f"field {field!r} is null — cannot evaluate rule")
val = self._coerce_value(val_str)
if isinstance(actual, str):
actual = self._coerce_value(actual)
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)
@staticmethod
def _coerce_value(value: str) -> Any:
"""Parse a rule literal into its most specific Python type."""
text = value.strip()
lowered = text.lower()
if lowered in ("true", "1"):
return True
if lowered in ("false", "0"):
return False
try:
return int(text)
except ValueError:
pass
try:
return float(text)
except ValueError:
pass
return text
# When crewai is absent there is no BaseTool to provide the public
# ``run``/``arun`` entry points, so expose them directly. With crewai
# installed these are left untouched so crewai's own implementations
# (usage tracking, ``result_as_answer``) win.
if not CREWAI_AVAILABLE:
def run(self, *args: Any, **kwargs: Any) -> str:
"""Run the tool synchronously (degraded mode, no crewai)."""
return self._run(*args, **kwargs)
async def arun(self, *args: Any, **kwargs: Any) -> str:
"""Run the tool asynchronously (degraded mode, no crewai)."""
return self._run(*args, **kwargs)
integrations/crewai/kg_tool.py
+573
View File
@@ -0,0 +1,573 @@
"""
SemanticaKGTool — a CrewAI ``BaseTool`` exposing Semantica's knowledge-graph
pipeline (``NERExtractor``, ``RelationExtractor``, ``ContextGraph``) to agents.
Lets agents build and query a shared ``ContextGraph`` as part of their
reasoning loop.
Install
-------
pip install semantica[crewai]
Example
-------
>>> from integrations.crewai import SemanticaKGTool
>>> from semantica.context import ContextGraph
>>> from crewai import Agent, Crew, Task
>>> graph = ContextGraph()
>>> tool = SemanticaKGTool(graph=graph)
>>> crew = Crew(
... agents=[Agent(role="...", goal="...", backstory="...", tools=[tool])],
... tasks=[...],
... )
Tools exposed
-------------
extract_entities — Extract named entities from text
extract_relations — Extract relationships between entities
add_to_graph — Extract entities/relations from text and add them to the graph
query_graph — Query the graph by keyword
find_related — Find concepts related to a given entity within ``hops``
"""
from __future__ import annotations
import json
import threading
import weakref
from typing import Any, Dict, List, Literal, Optional, Sequence, Type
from pydantic import BaseModel, Field
from semantica.utils.logging import get_logger
from ._availability import CREWAI_AVAILABLE, CREWAI_IMPORT_ERROR # noqa: F401
logger = get_logger(__name__)
# ---------------------------------------------------------------------------
# Optional: CrewAI BaseTool base class
# ---------------------------------------------------------------------------
_BaseTool: Any = object
if CREWAI_AVAILABLE:
from crewai.tools import BaseTool as _BaseTool # type: ignore
# One re-entrant lock per graph so concurrent tool invocations sharing a graph
# cannot double-count duplicate adds (check-then-act is not atomic), while
# independent graphs are never serialised against each other. An RLock also
# means an extractor callback that re-enters add_to_graph on the same graph
# cannot deadlock.
_graph_locks_guard = threading.Lock()
_graph_locks: "weakref.WeakKeyDictionary[Any, threading.RLock]" = (
weakref.WeakKeyDictionary()
)
# ---------------------------------------------------------------------------
# Input schema
# ---------------------------------------------------------------------------
class SemanticaKGToolInput(BaseModel):
"""
Input schema for ``SemanticaKGTool``.
Exactly one action is dispatched per call; the remaining fields are only
used by the actions that need them.
"""
action: Literal[
"extract_entities",
"extract_relations",
"add_to_graph",
"query_graph",
"find_related",
] = Field(
...,
description=(
"Which graph operation to run. One of: 'extract_entities', "
"'extract_relations', 'add_to_graph', 'query_graph', 'find_related'."
),
)
text: Optional[str] = Field(
None,
description=(
"Input text. Used by 'extract_entities', 'extract_relations' and "
"'add_to_graph'."
),
)
query: Optional[str] = Field(
None, description="Search query. Used by 'query_graph'."
)
entity: Optional[str] = Field(
None,
description="Root entity name. Used by 'find_related'.",
)
hops: int = Field(
1,
ge=1,
le=10,
description="Maximum relationship hops. Used by 'find_related'.",
)
# ---------------------------------------------------------------------------
# SemanticaKGTool
# ---------------------------------------------------------------------------
class SemanticaKGTool(_BaseTool): # type: ignore[misc]
"""
CrewAI tool that surfaces Semantica's KG pipeline as agent actions.
Parameters
----------
graph:
A ``semantica.context.ContextGraph`` to read/write. A fresh in-memory
graph is used when ``None``.
ner_extractor:
A ``semantica.semantic_extract.NERExtractor`` instance; auto-created
when ``None``.
relation_extractor:
A ``semantica.semantic_extract.RelationExtractor`` instance; auto-
created when ``None``.
"""
name: str = "semantica_knowledge_graph"
description: str = (
"Build and query a semantic knowledge graph. Actions: "
"'extract_entities' (extract named entities from 'text'), "
"'extract_relations' (extract relationships from 'text'), "
"'add_to_graph' (extract entities/relations from 'text' and add them "
"to the shared graph), 'query_graph' (keyword search using 'query'), "
"'find_related' (find concepts related to 'entity' within 'hops' "
"hops). Returns JSON."
)
args_schema: Type[BaseModel] = SemanticaKGToolInput
graph: Any = Field(default=None, exclude=True)
ner_extractor: Any = Field(default=None, exclude=True)
relation_extractor: Any = Field(default=None, exclude=True)
had_live_state: bool = False
reconstructed_state: bool = Field(default=False, exclude=True)
def __init__(
self,
graph: Any = None,
ner_extractor: Any = None,
relation_extractor: Any = None,
**kwargs: Any,
) -> None:
if CREWAI_AVAILABLE:
super().__init__(
graph=graph,
ner_extractor=ner_extractor,
relation_extractor=relation_extractor,
**kwargs,
)
else:
super().__init__()
self.graph = graph
self.ner_extractor = ner_extractor
self.relation_extractor = relation_extractor
# Degraded mode is a plain class — no model_post_init lifecycle.
self._ensure_defaults()
logger.info("SemanticaKGTool initialised (crewai=%s)", CREWAI_AVAILABLE)
def model_post_init(self, __context: Any) -> None:
"""Re-create default state after validation/deserialisation.
``graph``/extractors are excluded from JSON serialisation (CrewAI
checkpoints serialise every tool via ``model_dump(mode="json")``), so a
tool restored from a checkpoint has ``None`` state until this runs.
"""
self._ensure_defaults()
super().model_post_init(__context)
def _ensure_defaults(self) -> None:
"""Lazy-import and build defaults for any missing shared state."""
# Lazy imports keep the module importable without heavy deps
if self.graph is None:
from semantica.context import ContextGraph
self.graph = ContextGraph()
if self.had_live_state:
self.reconstructed_state = True
logger.warning(
"SemanticaKGTool: the live graph was lost during "
"serialization/checkpoint restore — an EMPTY graph was "
"reconstructed; re-attach the original graph before "
"continuing"
)
else:
logger.warning(
"SemanticaKGTool created a fresh in-memory ContextGraph — "
"agents sharing this tool's graph must be wired explicitly"
)
self.had_live_state = True
if self.ner_extractor is None:
from semantica.semantic_extract import NERExtractor
self.ner_extractor = NERExtractor()
if self.relation_extractor is None:
from semantica.semantic_extract import RelationExtractor
self.relation_extractor = RelationExtractor()
# ------------------------------------------------------------------
# CrewAI entry points
# ------------------------------------------------------------------
def _run(
self,
action: str,
text: Optional[str] = None,
query: Optional[str] = None,
entity: Optional[str] = None,
hops: int = 1,
**kwargs: Any,
) -> str:
"""
Dispatch a graph action. Always returns a JSON string so the agent
receives a structured, parseable result.
"""
valid = {
"extract_entities",
"extract_relations",
"add_to_graph",
"query_graph",
"find_related",
}
if action not in valid:
return json.dumps(
{
"error": f"Unknown action '{action}'. Valid actions: "
+ ", ".join(sorted(valid))
}
)
if action == "extract_entities":
return self._extract_entities(text or "")
if action == "extract_relations":
return self._extract_relations(text or "")
if action == "add_to_graph":
return self._add_from_text(text or "")
if action == "query_graph":
return self._query_graph(query or "")
return self._find_related(entity or "", hops=hops)
async def _arun(
self,
action: str,
text: Optional[str] = None,
query: Optional[str] = None,
entity: Optional[str] = None,
hops: int = 1,
**kwargs: Any,
) -> str:
"""
Async variant of ``_run`` for CrewAI's async tool path.
"""
return self._run(
action=action, text=text, query=query, entity=entity, hops=hops, **kwargs
)
# ------------------------------------------------------------------
# Entity/relation field access (handles both Semantica dataclasses and
# third-party shapes like MagicMock/plain dicts in stubs)
# ------------------------------------------------------------------
@staticmethod
def _first_str(obj: Any, attrs: Sequence[str]) -> str:
"""Return the first attribute value that is a non-empty string."""
for attr in attrs:
value = getattr(obj, attr, None)
if isinstance(value, str) and value:
return value
if isinstance(obj, dict):
for key in attrs:
value = obj.get(key)
if isinstance(value, str) and value:
return value
return ""
@classmethod
def _entity_name(cls, e: Any) -> str:
"""Best-effort name for an entity-like object."""
return cls._first_str(e, ("name", "text", "label", "node_id", "id"))
@classmethod
def _entity_type(cls, e: Any) -> str:
"""Best-effort type/label for an entity-like object."""
return cls._first_str(e, ("type", "label")) or "Entity"
@classmethod
def _relation_source(cls, r: Any) -> str:
"""Best-effort source of a relation-like object."""
src = cls._first_str(r, ("source",))
if not src:
src = cls._entity_name(getattr(r, "subject", None))
return src
@classmethod
def _relation_target(cls, r: Any) -> str:
"""Best-effort target of a relation-like object."""
tgt = cls._first_str(r, ("target",))
if not tgt:
tgt = cls._entity_name(getattr(r, "object", None))
return tgt
@classmethod
def _relation_type(cls, r: Any) -> str:
"""Best-effort relation type of a relation-like object."""
rtype = cls._first_str(r, ("type", "relation", "predicate"))
return rtype or "related_to"
@classmethod
def _confidence(cls, e: Any) -> float:
"""Normalise an entity/relation confidence value to a float."""
try:
val = getattr(e, "confidence", None)
if val is None:
return 1.0
return round(float(val), 4)
except (TypeError, ValueError):
return 1.0
@classmethod
def _graph_lock(cls, graph: Any) -> threading.RLock:
"""Return the re-entrant lock guarding a specific graph."""
with _graph_locks_guard:
lock = _graph_locks.get(graph)
if lock is None:
lock = threading.RLock()
_graph_locks[graph] = lock
return lock
# ------------------------------------------------------------------
# Actions
# ------------------------------------------------------------------
def _extract_entities(self, text: str) -> str:
"""Extract named entities from ``text``."""
try:
raw = self.ner_extractor.extract_entities(text) or []
entities = [
{
"name": self._entity_name(e),
"type": self._entity_type(e),
"confidence": self._confidence(e),
}
for e in raw
if self._entity_name(e)
]
logger.debug("extract_entities → %d entities", len(entities))
return json.dumps({"entities": entities, "count": len(entities)})
except Exception as exc:
logger.warning("extract_entities failed: %s", exc)
return json.dumps({"entities": [], "count": 0, "error": str(exc)})
def _extract_relations(self, text: str) -> str:
"""Extract relationships between entities in ``text``."""
try:
raw = self.relation_extractor.extract_relations(text) or []
relations = [
{
"source": self._relation_source(r),
"relation": self._relation_type(r),
"target": self._relation_target(r),
"confidence": self._confidence(r),
}
for r in raw
]
logger.debug("extract_relations → %d relations", len(relations))
return json.dumps({"relations": relations, "count": len(relations)})
except Exception as exc:
logger.warning("extract_relations failed: %s", exc)
return json.dumps({"relations": [], "count": 0, "error": str(exc)})
def _add_from_text(self, text: str) -> str:
"""
Extract entities and relations from ``text`` and add them to the graph.
Duplicate nodes/edges (same id, or same source/type/target) are
skipped so repeated calls are idempotent. Returns JSON with the
number of nodes/edges added.
"""
nodes_added = 0
edges_added = 0
try:
with self._graph_lock(self.graph):
existing_nodes = {
n.get("id") or n.get("node_id")
for n in (
self.graph.find_nodes() or [] # type: ignore[attr-defined]
)
if n.get("id") or n.get("node_id")
}
existing_edges = {
(e.get("source"), e.get("type") or "related_to", e.get("target"))
for e in (
self.graph.find_edges() or [] # type: ignore[attr-defined]
)
if e.get("source") and e.get("target")
}
raw_entities = self.ner_extractor.extract_entities(text) or []
entities: List[Any] = []
seen: set = set()
for e in raw_entities:
name = self._entity_name(e)
ntype = self._entity_type(e)
if not name or name in seen:
continue
seen.add(name)
entities.append(e)
if name in existing_nodes:
continue
try:
if self.graph.add_node(node_id=name, node_type=ntype):
nodes_added += 1
existing_nodes.add(name)
except Exception as exc:
logger.debug("add_node(%r) failed: %s", name, exc)
raw_relations = (
self.relation_extractor.extract_relations(text, entities=entities)
or []
)
for r in raw_relations:
src = self._relation_source(r)
tgt = self._relation_target(r)
rtype = self._relation_type(r)
if not src or not tgt:
continue
key = (src, rtype, tgt)
if key in existing_edges:
continue
try:
if self.graph.add_edge(
source_id=src, target_id=tgt, edge_type=rtype
):
edges_added += 1
existing_edges.add(key)
except Exception as exc:
logger.debug("add_edge(%r) failed: %s", key, exc)
logger.debug("add_to_graph: +%d nodes, +%d edges", nodes_added, edges_added)
return json.dumps({"nodes_added": nodes_added, "edges_added": edges_added})
except Exception as exc:
logger.warning("add_to_graph failed: %s", exc)
return json.dumps({"nodes_added": 0, "edges_added": 0, "error": str(exc)})
def _query_graph(self, query: str) -> str:
"""Keyword-search graph nodes by id, type and content."""
try:
q = (query or "").strip().lower()
out: List[dict] = []
seen: set = set()
query_method = getattr(self.graph, "query", None)
if query_method is not None:
for match in query_method(query) or []:
node = match.get("node") or {}
nid = node.get("id", "") or node.get("node_id", "")
if not nid or nid in seen:
continue
seen.add(nid)
content = match.get("content") or node.get("content", "")
out.append(
{
"id": nid,
"type": node.get("type", "") or node.get("node_type", ""),
"label": nid,
"content": str(content)[:500],
"score": round(float(match.get("score") or 0.0), 4),
}
)
if q:
for n in self.graph.find_nodes() or []: # type: ignore[attr-defined]
if isinstance(n, dict):
nid = n.get("id", "") or n.get("node_id", "")
ntype = n.get("type", "") or n.get("node_type", "")
content = str(
n.get("content")
or (n.get("properties") or {}).get("content", "")
or ""
)
else:
nid = getattr(n, "id", getattr(n, "label", ""))
ntype = getattr(n, "node_type", "")
content = str(getattr(n, "content", "") or "")
if not nid or nid in seen:
continue
if q in str(nid).lower() or q in str(ntype).lower():
seen.add(nid)
out.append(
{
"id": nid,
"type": ntype,
"label": nid,
"content": content[:500],
"score": 1.0,
}
)
return json.dumps({"results": out, "count": len(out)})
except Exception as exc:
logger.warning("query_graph failed: %s", exc)
return json.dumps({"results": [], "count": 0, "error": str(exc)})
def _find_related(self, entity: str, hops: int = 1) -> str:
"""Find concepts related to ``entity`` within ``hops`` graph hops.
Traversal is undirected — an edge counts as related regardless of
direction, so both outgoing and incoming edges are honored.
"""
try:
adjacency: Dict[str, List[str]] = {}
for edge in self.graph.find_edges() or []: # type: ignore[attr-defined]
if isinstance(edge, dict):
src = edge.get("source")
tgt = edge.get("target")
else:
src = getattr(edge, "source", None)
tgt = getattr(edge, "target", None)
if not src or not tgt:
continue
adjacency.setdefault(src, []).append(tgt)
adjacency.setdefault(tgt, []).append(src)
related: List[str] = []
frontier = [entity]
visited = {entity}
for _ in range(max(1, hops)):
next_frontier: List[str] = []
for e in frontier:
for n in adjacency.get(e, []):
if n in visited:
continue
visited.add(n)
next_frontier.append(n)
related.append(n)
frontier = next_frontier
logger.debug("find_related('%s', hops=%d) → %d", entity, hops, len(related))
return json.dumps(
{"entity": entity, "related": related, "count": len(related)}
)
except Exception as exc:
logger.warning("find_related failed: %s", exc)
return json.dumps(
{"entity": entity, "related": [], "count": 0, "error": str(exc)}
)
# When crewai is absent there is no BaseTool to provide the public
# ``run``/``arun`` entry points, so expose them directly. With crewai
# installed these are left untouched so crewai's own implementations
# (usage tracking, ``result_as_answer``) win.
if not CREWAI_AVAILABLE:
def run(self, *args: Any, **kwargs: Any) -> str:
"""Run the tool synchronously (degraded mode, no crewai)."""
return self._run(*args, **kwargs)
async def arun(self, *args: Any, **kwargs: Any) -> str:
"""Run the tool asynchronously (degraded mode, no crewai)."""
return self._run(*args, **kwargs)
integrations/crewai/knowledge_source.py
+331
View File
@@ -0,0 +1,331 @@
"""
SemanticaKnowledgeSource — expose a Semantica ``ContextGraph`` as a CrewAI
knowledge source.
Lets a ``Crew`` load the current state of a knowledge graph (nodes, edges,
metadata) into its knowledge storage, so every agent gets retrieval access to
graph knowledge during the kickoff.
Install
-------
pip install semantica[crewai]
Example
-------
>>> from integrations.crewai import SemanticaKnowledgeSource
>>> from semantica.context import ContextGraph
>>> from crewai import Agent, Crew, Task
>>> graph = ContextGraph()
>>> graph.add_node(node_id="privacy", node_type="policy")
>>> crew = Crew(
... agents=[...],
... tasks=[...],
... knowledge_sources=[SemanticaKnowledgeSource(graph=graph)],
... )
Compatibility
-------------
Works with ``crewai >= 0.80.0``. The ``BaseKnowledgeSource`` contract changed
between versions (``load_content`` → ``validate_content``/``aadd``), so this
source implements both legacy and current methods. It degrades gracefully
when ``crewai`` is not installed: the class is still importable and carries the
full Semantica API, but cannot be passed to a ``Crew``.
"""
from __future__ import annotations
import asyncio
from typing import Any, Dict, List, Optional
from pydantic import Field
from semantica.utils.logging import get_logger
from ._availability import CREWAI_AVAILABLE
logger = get_logger(__name__)
# ---------------------------------------------------------------------------
# Optional: CrewAI BaseKnowledgeSource base class
# ---------------------------------------------------------------------------
_BaseKnowledgeSource: Any = object
if CREWAI_AVAILABLE:
from crewai.knowledge.source.base_knowledge_source import (
BaseKnowledgeSource as _BaseKnowledgeSource, # type: ignore
)
def _chunk_text_manual(text: str, chunk_size: int, chunk_overlap: int) -> List[str]:
"""Fallback plain-text chunker for when CrewAI helpers are unavailable."""
if not text:
return []
if int(chunk_size) <= 0:
return [text]
size = max(1, int(chunk_size))
overlap = max(0, int(chunk_overlap))
if len(text) <= size:
return [text]
step = max(1, size - overlap)
return [text[i : i + size] for i in range(0, len(text), step)]
class SemanticaKnowledgeSource(_BaseKnowledgeSource): # type: ignore[misc]
"""
CrewAI knowledge source backed by a Semantica ``ContextGraph``.
On ``add()`` the graph's nodes and edges are serialised into readable text
and pushed through the standard CrewAI chunking / storage pipeline, making
graph knowledge retrievable by every agent in the crew.
Parameters
----------
graph:
A ``semantica.context.ContextGraph`` to expose. A fresh in-memory
graph is created when ``None``.
name:
Source name. Defaults to ``"semantica_knowledge_graph"``.
chunk_size:
Max characters per chunk (default 4000).
chunk_overlap:
Character overlap between adjacent chunks (default 200).
"""
name: str = "semantica_knowledge_graph"
graph: Any = Field(default=None, exclude=True)
chunk_size: int = 4000
chunk_overlap: int = 200
had_live_state: bool = False
reconstructed_state: bool = Field(default=False, exclude=True)
def __init__(
self,
graph: Any = None,
name: Optional[str] = None,
chunk_size: int = 4000,
chunk_overlap: int = 200,
**kwargs: Any,
) -> None:
if CREWAI_AVAILABLE:
# Do NOT eagerly build a graph here: pydantic calls this ``__init__``
# during ``model_validate`` (checkpoint restore), and the eager
# build would hide that a live graph was lost. ``model_post_init``
# rebuilds defaults and flags ``reconstructed_state`` instead.
super().__init__(
graph=graph,
name=name or "semantica_knowledge_graph",
chunk_size=int(chunk_size),
chunk_overlap=int(chunk_overlap),
**kwargs,
)
else:
if graph is None:
from semantica.context import ContextGraph
graph = ContextGraph()
super().__init__()
self.graph = graph
self.name = name or "semantica_knowledge_graph"
self.chunk_size = int(chunk_size)
self.chunk_overlap = int(chunk_overlap)
logger.info(
"SemanticaKnowledgeSource initialised (crewai=%s, chunk_size=%d)",
CREWAI_AVAILABLE,
self.chunk_size,
)
self.had_live_state = True
def model_post_init(self, __context: Any) -> None:
"""Re-create default state after validation/deserialisation.
``graph`` is excluded from JSON serialisation (CrewAI checkpoints
serialise their models via ``model_dump(mode="json")``), so a source
restored from a checkpoint has ``None`` state until this runs.
"""
if self.graph is None:
from semantica.context import ContextGraph
self.graph = ContextGraph()
if self.had_live_state:
self.reconstructed_state = True
logger.warning(
"SemanticaKnowledgeSource: the live graph was lost during "
"serialization/checkpoint restore — an EMPTY graph was "
"reconstructed; re-attach the original graph before "
"continuing"
)
else:
logger.warning(
"SemanticaKnowledgeSource created a fresh in-memory "
"ContextGraph — sources sharing knowledge must be wired to "
"the same graph explicitly"
)
self.had_live_state = True
super().model_post_init(__context)
# ------------------------------------------------------------------
# Content extraction
# ------------------------------------------------------------------
def load_content(self) -> Dict[str, str]:
"""
Serialise the graph into ``{id: readable_text}`` pairs.
Nodes are rendered with their type/content/metadata, edges with their
source, relation type and target. This satisfies the legacy CrewAI
``BaseKnowledgeSource.load_content`` contract.
"""
content: Dict[str, str] = {}
graph = self.graph
if graph is None:
return content
try:
for node in graph.find_nodes() or []: # type: ignore[attr-defined]
nid = node.get("id") or node.get("node_id") or ""
if not nid:
continue
parts = [
"Entity",
str(nid),
"type: " + str(node.get("type", "entity")),
]
if node.get("content"):
parts.append("content: " + str(node["content"]))
if node.get("metadata"):
try:
import json
parts.append("metadata: " + json.dumps(node["metadata"]))
except Exception:
parts.append("metadata: " + str(node["metadata"]))
content[str(nid)] = " | ".join(parts)
except Exception as exc:
logger.warning(
"SemanticaKnowledgeSource.load_content (nodes) failed: %s", exc
)
try:
for idx, edge in enumerate(
graph.find_edges() or [] # type: ignore[attr-defined]
):
src = edge.get("source")
tgt = edge.get("target")
if not src or not tgt:
continue
rel = edge.get("type") or edge.get("edge_type") or "related_to"
weight = edge.get("weight")
text = f"{src} -[{rel}]-> {tgt}"
if weight is not None:
text += f" (weight: {weight})"
content[f"edge-{idx}"] = text
except Exception as exc:
logger.warning(
"SemanticaKnowledgeSource.load_content (edges) failed: %s", exc
)
return content
def validate_content(self) -> Any:
"""
Validate that a readable graph is attached.
Satisfies the current CrewAI ``BaseKnowledgeSource.validate_content``
contract.
"""
if self.graph is None:
raise ValueError("SemanticaKnowledgeSource requires a ContextGraph.")
return True
# ------------------------------------------------------------------
# Chunking + storage (abstract in both CrewAI generations)
# ------------------------------------------------------------------
def _chunk(self, text: str) -> List[str]:
"""Chunk ``text`` using CrewAI's helper when available, else manual."""
helper = getattr(self, "_chunk_text", None)
if helper is not None:
try:
return list(helper(text) or [])
except Exception as exc:
logger.debug(
"SemanticaKnowledgeSource._chunk_text failed, falling back: %s", exc
)
return _chunk_text_manual(text, self.chunk_size, self.chunk_overlap)
def add(self) -> None:
"""
Process the graph into chunks and store them via CrewAI storage.
Sets both ``chunks`` (current CrewAI) and ``_chunks`` (legacy CrewAI)
so either ``_save_documents`` implementation picks them up. If no
storage has been wired (e.g. not yet attached to a ``Crew``), chunks
are kept in memory.
"""
content = self.load_content()
if not content:
logger.debug("SemanticaKnowledgeSource.add: empty graph — nothing to store")
return
chunks: List[str] = []
for _, text in content.items():
if text:
chunks.extend(self._chunk(text))
self.chunks = chunks
self._chunks = chunks
save = getattr(self, "_save_documents", None)
if save is not None:
if getattr(self, "storage", None) is None:
logger.debug(
"SemanticaKnowledgeSource.add: storage not wired — "
"keeping chunks in memory"
)
else:
try:
save()
logger.info(
"SemanticaKnowledgeSource.add: stored %d chunks", len(chunks)
)
return
except Exception as exc:
logger.error(
"SemanticaKnowledgeSource.add: storage save FAILED (%s) — "
"chunks are only kept in memory and agents will retrieve "
"nothing. Configure the Crew embedder (e.g. an OpenAI "
"embedder with OPENAI_API_KEY, or a local embedder) before "
"running the crew.",
exc,
)
logger.info(
"SemanticaKnowledgeSource.add: %d chunks ready in memory", len(chunks)
)
async def aadd(self) -> None:
"""
Asynchronous variant of ``add()`` (current CrewAI contract).
The graph serialisation is CPU-bound, so it runs in a thread pool to
avoid blocking the event loop.
"""
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self.add)
# ------------------------------------------------------------------
# Inspection helpers
# ------------------------------------------------------------------
def get_content_summary(self) -> Dict[str, Any]:
"""
Summarise what the source exposes (helpful for debugging / testing).
"""
content = self.load_content()
return {
"name": self.name,
"source_count": len(content),
"chunks": len(getattr(self, "chunks", []) or []),
"crewai_available": CREWAI_AVAILABLE,
}