Multi-agent orchestration pipeline for autonomous enterprise trend intelligence — research, risk-assess, and generate production-grade content.
Quick Start • Architecture • API • Configuration • Structure • Docker
Enterprise Intelligence Crew is a preemptive, stateful, multi-layer orchestration system that breaks the standard CrewAI crew.kickoff() monolith into three isolated mini-crew executions, each validated by a gate before data flows downstream. All LLM inference runs locally via Ollama — no API keys, no cloud dependency, no data egress. A dual-gate risk validation system enforces quality guardrails between stages, with feedback injection and circuit-breaker retry logic. ChromaDB with sentence transformers persists research for semantic recall.
💡 Verdict: This architecture trades simplicity and speed for correctness, auditability, and local privacy.
Ideal for:
- Regulated environments (healthcare, finance) where data cannot leave the network
- High-stakes content generation where a bad output has real cost (compliance, brand risk)
- Teams that value testability and need CI to run without LLM infrastructure
Over-engineered for:
- Rapid prototyping where a standard
crew.kickoff()is sufficient - High-throughput, low-latency use cases (gate overhead and sequential mini-crews are bottlenecks)
- Multi-instance deployments (file-based task store and local ChromaDB don't scale horizontally)
- Python 3.12+
- Ollama — with at least one model pulled:
ollama pull qwen2.5:1.5b ollama serve
- No cloud API keys required.
git clone https://github.com/aragit/enterprise-intelligence-crew.git
cd enterprise-intelligence-crew
pip install -r requirements.txt
# CLI mode
python main.py
# API server
python3 -m uvicorn src.api.main:app --host 0.0.0.0 --port 8000
# With mock LLM (no Ollama needed — for testing)
LLM_PROVIDER=mock python main.py
# Run tests
python3 -m pytest tests/ -vPreemptive mini-crew orchestration. Instead of a single crew.kickoff() monolith, the pipeline runs three isolated 1-agent, 1-task mini-crews sequentially. Each execution is validated by a gate before data flows downstream, preventing malformed outputs from poisoning downstream agents.
| Stage | Agent | Tools | Output | Gate |
|---|---|---|---|---|
| 1 | TrendInvestigator | WebSearch, WebScraper, SentimentAnalyzer | TrendPayload |
Gate 1 — Pure Python validator (<1ms) |
| 2 | RiskAnalyst | BiasDetector, Validator | RiskPayload |
Gate 2 — LangGraph StateGraph (5-node compiled graph) |
| 3 | Copywriter | SEOOptimizer, Summarizer | ContentPayload |
— |
-
Dual-gate validation. Gate 1 checks TrendPayload bounds (momentum_score 0.0–1.0), non-empty sources, and metrics. Gate 2 evaluates risk_score against threshold 0.7 via a compiled LangGraph loop (analyze → evaluate_risk → [approve|reject]). Both gates support feedback injection: rejected outputs trigger agent rerun with contextual feedback, up to max_iterations=3 before circuit-breaker force-approval.
-
Schema-aware extraction. _extract_payload() implements a 4-tier fallback (direct Pydantic → CrewOutput traversal → JSON string → raw dict) to handle output format drift across LLM providers.
-
Memory. CrewMemory replaces CrewAI's internal memory (which defaults to OpenAI extraction) with ChromaDB + all-MiniLM-L6-v2 embeddings — fully local, zero API keys.
-
API surface. FastAPI with 6 endpoints (/health, /crew/run, /crew/run/async, /crew/status/{id}, /crew/memory/{topic}, /metrics). Task state persisted via fcntl-locked JSON. Prometheus instrumentation and Loguru rotation included.
Advantages
| # | Advantage | Why It Matters |
|---|---|---|
| 1 | Fault Isolation | Each agent runs in its own mini-crew. A hallucinated TrendPayload cannot silently poison the RiskAnalyst or Copywriter. |
| 2 | Data Integrity by Contract | Pydantic gates enforce TrendPayload/RiskPayload structure before downstream execution. Bad data is caught, not propagated. |
| 3 | Provider Agnostic | OllamaNativeLLM + OpenAIProvider + MockNativeLLM — swap providers via env var without touching agent code. |
| 4 | Resilient to Output Drift | 4-tier _extract_payload() handles format variations across Ollama, OpenAI, and CrewAI versions. |
| 5 | Auditable State Machine | PipelineState is explicit and mutable. Every feedback injection and retry is logged and traceable — unlike CrewAI's hidden internal memory. |
| 6 | Autonomous Retry with Safety | Feedback loops + circuit breaker (max_iterations=3) let the system self-correct without infinite loops or human intervention. |
| 7 | No Vendor Lock-in on Memory | Custom CrewMemory replaces CrewAI's default OpenAI-based memory extraction. ChromaDB + sentence-transformers is fully open-source. |
| 8 | Cost Control | Local inference eliminates per-token cloud LLM costs. |
| Tool | Library | What It Does |
|---|---|---|
web_search |
duckduckgo_search |
DDGS text search, returns title/href/body |
web_scraper |
trafilatura + requests |
HTML extraction, title, word count |
sentiment_analyzer |
textblob |
Polarity (-1 to 1), subjectivity, label |
bias_detector |
Heuristic | 4-category keyword scan (hyperbolic, emotional, certainty, vagueness) + source domain diversity score |
validator |
urllib.parse + regex |
TLD trust (.edu, .gov, .org), known-credible domain whitelist, content length check |
seo_optimizer |
textstat |
Flesch readability, grade level, keyword density, optimization suggestions |
summarizer |
sumy (LSA) |
Extractive summarization, configurable sentence count |
crew_tools |
crewai |
7 BaseTool wrappers with TOOL_REGISTRY mapping |
All tools wrapped with @retry(max_attempts=3, delay=1.0) and return ToolResult(success, data, error, duration_ms).
Every stage produces a strictly validated payload:
| Contract | Fields | Validation |
|---|---|---|
TrendPayload |
trend_name, momentum_score (0.0–1.0), extracted_metrics (dict), verified_sources (list[str]) |
Regex URL validator (http(s):// + valid host); rejects localhost, bare IPs; extracts URLs from raw text |
RiskPayload |
is_safe (bool), risk_score (0.0–1.0), flagged_keywords (list), required_revisions (list) |
— |
ContentPayload |
headline, body_content, metadata_tags (list[str]) |
— |
All payloads use str URLs (not HttpUrl) to avoid json.dumps() serialization issues in CrewAI task storage. URLs are validated via regex with extraction from raw text and placeholder fallback. Each payload has __version__ = "1.0".
CrewAI-Compatible Adapters:
OllamaNativeLLM— Extendscrewai.llms.base_llm.BaseLLM, calls Ollama's native/api/chatdirectly viahttpx. Bypasses CrewAI'sOpenAICompatibleCompletion(which requires/v1/chat/completions, unavailable on Ollama ≤0.24). Supports tool calling, structured output, stop words, streaming, and usage extraction. Context window: 131,072 tokens.MockNativeLLM— Returns schema-valid JSON forresponse_model. Enables the full 76-test suite without any LLM connection.
Standalone Providers (for direct use outside CrewAI):
BaseLLMProvider(ABC) →OllamaProvider,OpenAIProvider,MockProvider
Provider selection is controlled by LLM_PROVIDER env var (ollama | openai | mock).
CrewMemory replaces CrewAI's internal memory (which defaults to OpenAI gpt-4o-mini extraction, violating the local-only constraint):
- Backend:
chromadb.PersistentClientwithhnsw:space = "cosine", collectionenterprise_research - Embeddings:
sentence-transformers/all-MiniLM-L6-v2with a_FallbackEmbedder(384-dim word-hash) if the dependency is missing - API: FastAPI endpoint
/crew/memory/{topic}performs semantic search - Storage: Full pipeline result is embedded and persisted after Copywriter completion
| Method | Path | Description |
|---|---|---|
GET |
/health |
Provider status, model, health message |
POST |
/crew/run |
Execute pipeline (synchronous) |
POST |
/crew/run/async |
Execute pipeline (async, returns task_id) |
GET |
/crew/status/{task_id} |
Poll async result |
GET |
/crew/memory/{topic} |
ChromaDB semantic search |
GET |
/metrics |
Prometheus scrape endpoint |
curl -s -X POST http://localhost:8000/crew/run \
-H "Content-Type: application/json" \
-d '{"query_context": "Edge AI adoption in healthcare 2026"}' | jq .Pydantic Settings with .env override:
| Variable | Default | Options |
|---|---|---|
LLM_PROVIDER |
ollama |
ollama, openai, mock |
OLLAMA_BASE_URL |
http://localhost:11434 |
Any Ollama host |
OLLAMA_MODEL |
qwen2.5:1.5b |
Any pulled model |
OPENAI_API_KEY |
— | Required for openai |
OPENAI_MODEL |
gpt-4o-mini |
Any OpenAI model |
CHROMADB_PATH |
./data/chromadb |
Persistence path |
LOG_LEVEL |
INFO |
DEBUG, INFO, WARNING |
├── src/
│ ├── agents/
│ │ └── intelligence_crew.py # 3 mini-crew orchestration, PipelineState, _extract_payload
│ ├── api/
│ │ ├── main.py # FastAPI app, lifespan, instrumentation
│ │ └── routes.py # Endpoints: health, crew run, memory, metrics
│ ├── memory/
│ │ └── crew_memory.py # ChromaDB vector store client
│ ├── orchestration/
│ │ └── risk_gate.py # Dual-gate: pure Python (Gate 1) + LangGraph StateGraph (Gate 2)
│ ├── schemas/
│ │ └── payloads.py # Pydantic V2 contracts with URL sanitization + version markers
│ ├── tools/
│ │ ├── web_search.py # DuckDuckGo search
│ │ ├── web_scraper.py # Trafilatura HTML extraction
│ │ ├── summarizer.py # LSA extractive summary (sumy)
│ │ ├── sentiment_analyzer.py # TextBlob sentiment
│ │ ├── bias_detector.py # Heuristic bias scan
│ │ ├── validator.py # Source credibility scoring
│ │ ├── seo_optimizer.py # Readability, keyword density
│ │ └── crew_tools.py # 7 CrewAI BaseTool wrappers + TOOL_REGISTRY
│ ├── llm.py # OllamaNativeLLM, MockNativeLLM, BaseLLMProvider hierarchy
│ └── config.py # Pydantic Settings with env overrides
├── configs/
│ └── crew_config.yaml # Agent roles, goals, backstories (+ {feedback} placeholder)
├── tests/ # 76 tests (pytest, 1 skipped — web search needs network)
│ ├── test_gates.py # Gate 1 + Gate 2 logic + consistency
│ ├── test_payload_extraction.py # 4-tier _extract_payload validation
│ ├── test_pipeline_state.py # PipelineState serialization
│ ├── test_feedback_injection.py # Feedback loop + sanitization
│ ├── test_dependencies.py # Dead dependency verification
│ ├── test_mini_crew.py # Per-agent mini-crew execution
│ ├── test_pipeline_integration.py # Full pipeline + circuit breaker
│ ├── test_gate_consistency.py # Cross-gate drift detection
│ ├── test_schema_e2e.py # URL sanitization + bounds
│ ├── test_performance.py # Latency benchmarks
│ ├── test_concurrency.py # Thread safety + file locks
│ ├── test_security.py # Prompt injection + malformed payloads
│ └── test_functional.py # Subprocess-based functional test
├── AGENTS.md # Anchored summary — architecture decisions + test strategy
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
└── requirements.txt # langchain + langchain-chroma REMOVED
docker compose up --buildPulls the model inside the container:
docker compose exec ollama ollama pull qwen2.5:1.5b# Full suite (76 tests, 1 skipped — web search requires network)
pytest tests/ -v
# With coverage
pytest tests/ --cov=src --cov-report=term-missing
# Specific test categories
pytest tests/test_gates.py -v # Gate logic
pytest tests/test_payload_extraction.py -v # Schema extraction
pytest tests/test_feedback_injection.py -v # Feedback loop
pytest tests/test_performance.py -v # Benchmarks
pytest tests/test_security.py -v # Security hardening| Decision | Rationale |
|---|---|
| Native Ollama adapter | CrewAI's OpenAICompatibleCompletion needs /v1/chat/completions (Ollama ≥0.28). OllamaNativeLLM uses /api/chat — compatible with all versions. |
| ChromaDB over LanceDB | Avoids CrewAI's internal memory which defaults to OpenAI for memory extraction. ChromaDB + sentence-transformers is fully local, no API keys. |
Agent memory=False |
CrewAI internal memory requires gpt-4o-mini extraction by default. Our own CrewMemory handles persistence. |
str URLs over HttpUrl |
Pydantic's HttpUrl breaks json.dumps() in CrewAI's task output storage. String URLs with a field_validator solve this. |
| No LiteLLM dependency | LiteLLM is optional in CrewAI 1.14. We use httpx directly — fewer dependencies, simpler debugging. |
| Preemptive dual-gate validation | Breaking crew.kickoff() into per-agent mini-crews with intermediate gates prevents bad data from flowing downstream. Gate 1 (pure Python) is fast; Gate 2 (LangGraph) is expressive. |
Schema-aware _extract_payload |
Replaced fragile string-matching (_parse_crew_output) with 4-tier Pydantic validation: direct instance → TaskOutput → JSON string → raw dict. Handles CrewAI output format variations across models. |
| Feedback injection via YAML placeholder | {feedback} in crew_config.yaml task descriptions allows runtime injection of gate feedback without hardcoding strings in Python. Enables autonomous retry loops. |
| Dead dependency removal | Removed langchain and langchain-chroma (zero imports in codebase). Reduced install time and attack surface. |
MIT