#4606 CrewAI and Ollama Board of Directors for Innovation
#4606 CrewAI and Ollama Board of Directors for Innovation
https://docs.crewai.com/v1.15.3/en/tools/overview
#4606 Provide a python app using CrewAI and Ollama supported local models with a board of directors to 1) find opportunities, 2) analyze the opportunity, 3) suggest solutions, and 4) implement solutions using locally implemented agents that support local tool use. Include a critic for each agent and feed back its response to the original agent for follow-up for each step. Support Web Search and File tools.
Here is a complete, production-ready Python application using CrewAI and Ollama local models.
To satisfy the requirement of having a Critic for each Agent that sends feedback back to the original agent for revisions before moving to the next phase, we use CrewAI Flows. Flows provide the state management and routing hooks necessary to create iterative critique loops between agents.
Core Prerequisites
First, make sure you have the required packages installed:
pip install crewai crewai-tools langchain-community
Make sure Ollama is running locally and you have a model pulled (we will use llama3.1 or qwen2.5 as they handle local tool calling and structured reasoning well):
ollama pull llama3.1
Complete Python Application (board_room.py)
import os
from typing import Dict, Any
from pydantic import BaseModel, Field
from crewai import Agent, Task, Crew, Process, LLM
from crewai.flow import Flow, listen, start, router
from crewai_tools import SerperDevTool, FileWriterTool
# 1. Initialize Local LLM via Ollama
# Ensure the model selected is highly capable of JSON/Tool formatting locally
local_llm = LLM(
model="ollama/llama3.1",
base_url="http://localhost:11434",
temperature=0.2
)
# 2. Setup Local Tools
# Note: For SerperDevTool, you'll need `os.environ["SERPER_API_KEY"]` or swap it out for a purely local scraper if completely offline.
web_search_tool = SerperDevTool()
file_writer_tool = FileWriterTool()
# ---------------------------------------------------------------------
# Pydantic Schemas for Structured States and Critique Outputs
# ---------------------------------------------------------------------
class CritiqueResult(BaseModel):
approved: bool = Field(description="True if the output meets all criteria, False if revisions are required.")
feedback: str = Field(description="Detailed constructive criticism or instructions for improvement.")
# ---------------------------------------------------------------------
# CrewAI Flow Architecture
# ---------------------------------------------------------------------
class BoardOfDirectorsFlow(Flow):
def __init__(self):
super().__init__()
# Initializing global state metrics to track revision loop counts
self.state["opportunity"] = ""
self.state["analysis"] = ""
self.state["solution"] = ""
self.state["implementation_plan"] = ""
self.state["loop_counts"] = {
"finder": 0,
"analyst": 0,
"strategist": 0,
"executor": 0
}
self.max_revisions = 2 # Guardrail against infinite local loops
# ==========================================
# PHASE 1: FIND OPPORTUNITY + CRITIC
# ==========================================
@start()
def find_opportunity(self):
print("\n=== [Phase 1] Finder Agent: Searching for Opportunities ===")
finder = Agent(
role="Opportunity Scout",
goal="Identify emerging market gaps, high-value technical trends, or business operational friction points.",
backstory="You excel at parsing massive data trends and pinpointing real-world opportunities.",
tools=[web_search_tool],
llm=local_llm,
verbose=True
)
description = "Find one major emerging opportunity in the AI/Automation sector for 2026."
if self.state["loop_counts"]["finder"] > 0:
description += f"\n\nYOUR PREVIOUS ATTEMPT WAS REJECTED. Incorporate this Board feedback:\n{self.state['finder_feedback']}"
task = Task(
description=description,
expected_output="A summary profiling a singular concrete business or market opportunity.",
agent=finder
)
crew = Crew(agents=[finder], tasks=[task], process=Process.sequential)
self.state["opportunity"] = str(crew.kickoff())
return "evaluate_opportunity"
@listen("evaluate_opportunity")
def evaluate_opportunity(self):
print("--- Board Critic: Auditing the Opportunity ---")
critic = Agent(
role="Opportunity Critic",
goal="Ruthlessly audit the proposed opportunity for viability, uniqueness, and immediate relevance.",
backstory="You are a skeptical director who rejects vague or oversaturated concepts.",
llm=local_llm
)
task = Task(
description=f"Review this proposed opportunity:\n\n{self.state['opportunity']}\n\nDetermine if it is valid or needs revision.",
expected_output="JSON structure matching the requested schema indicating approval status and feedback.",
output_json=CritiqueResult,
agent=critic
)
crew = Crew(agents=[critic], tasks=[task])
result = crew.kickoff()
# Enforce loop breaker caps
if self.state["loop_counts"]["finder"] >= self.max_revisions:
print("⚠️ Max revisions reached for Finder. Forcing approval.")
return "analyze_opportunity"
if not result["approved"]:
print(f"❌ Rejected by Critic. Feedback: {result['feedback']}")
self.state["finder_feedback"] = result["feedback"]
self.state["loop_counts"]["finder"] += 1
return "find_opportunity" # Routes back to Finder
print("✅ Opportunity Approved by Critic.")
return "analyze_opportunity"
# ==========================================
# PHASE 2: ANALYZE OPPORTUNITY + CRITIC
# ==========================================
@listen("analyze_opportunity")
def analyze_opportunity(self):
print("\n=== [Phase 2] Analyst Agent: Investigating the Target ===")
analyst = Agent(
role="Strategic Risk and Value Analyst",
goal="Perform a deep-dive analysis on the provided opportunity, detailing SWOT and economic viability.",
backstory="Data is your guiding star. You translate trends into concrete feasibility metrics.",
tools=[web_search_tool],
llm=local_llm,
verbose=True
)
description = f"Conduct a SWOT and financial feasibility analysis on this opportunity:\n\n{self.state['opportunity']}"
if self.state["loop_counts"]["analyst"] > 0:
description += f"\n\nREVISION REQUIRED. Address this Critic feedback:\n{self.state['analyst_feedback']}"
task = Task(
description=description,
expected_output="A structured report comprising SWOT analysis and competitive landscape.",
agent=analyst
)
crew = Crew(agents=[analyst], tasks=[task])
self.state["analysis"] = str(crew.kickoff())
return "evaluate_analysis"
@listen("evaluate_analysis")
def evaluate_analysis(self):
print("--- Board Critic: Auditing the Analysis ---")
critic = Agent(
role="Analysis Critic",
goal="Audit the strategic analysis report for logical blind spots, missed risks, or exaggerated ROI.",
backstory="You look for flaws in analytical math and challenge optimistic assumptions.",
llm=local_llm
)
task = Task(
description=f"Review this analysis:\n\n{self.state['analysis']}\n\nAssess if it is complete and sound.",
expected_output="JSON mapping to standard schema detailing approval and critique notes.",
output_json=CritiqueResult,
agent=critic
)
crew = Crew(agents=[critic], tasks=[task])
result = crew.kickoff()
if self.state["loop_counts"]["analyst"] >= self.max_revisions:
return "suggest_solutions"
if not result["approved"]:
print(f"❌ Rejected by Critic. Feedback: {result['feedback']}")
self.state["analyst_feedback"] = result["feedback"]
self.state["loop_counts"]["analyst"] += 1
return "analyze_opportunity"
print("✅ Analysis Approved by Critic.")
return "suggest_solutions"
# ==========================================
# PHASE 3: SUGGEST SOLUTIONS + CRITIC
# ==========================================
@listen("suggest_solutions")
def suggest_solutions(self):
print("\n=== [Phase 3] Strategist Agent: Designing Solutions ===")
strategist = Agent(
role="Solution Architect",
goal="Design actionable solutions and technical architectures that perfectly fit the opportunity analysis.",
backstory="You excel at transforming conceptual problem sets into tangible, concrete technical systems.",
llm=local_llm,
verbose=True
)
description = f"Based on the analysis:\n{self.state['analysis']}\n\nPropose 2 potential tactical solutions."
if self.state["loop_counts"]["strategist"] > 0:
description += f"\n\nREVISION REQUIRED. Address this Critic feedback:\n{self.state['strategist_feedback']}"
task = Task(
description=description,
expected_output="A blueprint outlining two actionable solutions with pros and cons.",
agent=strategist
)
crew = Crew(agents=[strategist], tasks=[task])
self.state["solution"] = str(crew.kickoff())
return "evaluate_solutions"
@listen("evaluate_solutions")
def evaluate_solutions(self):
print("--- Board Critic: Auditing the Solutions ---")
critic = Agent(
role="Solution Critic",
goal="Evaluate the proposed blueprints for architectural feasibility, cost, and alignment.",
backstory="You reject overly engineered solutions or plans that ignore operational reality.",
llm=local_llm
)
task = Task(
description=f"Review these solutions:\n\n{self.state['solution']}\n\nDetermine approval.",
expected_output="Structured JSON indicating approval state.",
output_json=CritiqueResult,
agent=critic
)
crew = Crew(agents=[critic], tasks=[task])
result = crew.kickoff()
if self.state["loop_counts"]["strategist"] >= self.max_revisions:
return "implement_solutions"
if not result["approved"]:
print(f"❌ Rejected by Critic. Feedback: {result['feedback']}")
self.state["strategist_feedback"] = result["feedback"]
self.state["loop_counts"]["strategist"] += 1
return "suggest_solutions"
print("✅ Solutions Approved by Critic.")
return "implement_solutions"
# ==========================================
# PHASE 4: IMPLEMENTATION + CRITIC
# ==========================================
@listen("implement_solutions")
def implement_solutions(self):
print("\n=== [Phase 4] Execution Agent: Building Implementation Steps ===")
executor = Agent(
role="Operations Director",
goal="Build a granular rollout plan and execute artifact deployment safely using local capabilities.",
backstory="You transform technical blueprints into operational reality and write the deployment files.",
tools=[file_writer_tool],
llm=local_llm,
verbose=True
)
description = (
f"Develop a step-by-step implementation plan for the chosen solution blueprint:\n{self.state['solution']}\n\n"
f"You MUST use the FileWriterTool to save this implementation plan directly to a file named 'implementation_plan.md'."
)
if self.state["loop_counts"]["executor"] > 0:
description += f"\n\nREVISION REQUIRED. Address this Critic feedback:\n{self.state['executor_feedback']}"
task = Task(
description=description,
expected_output="A deployment playbook saved to a local markdown file.",
agent=executor
)
crew = Crew(agents=[executor], tasks=[task])
self.state["implementation_plan"] = str(crew.kickoff())
return "evaluate_implementation"
@listen("evaluate_implementation")
def evaluate_implementation(self):
print("--- Board Critic: Auditing the Implementation & Artifacts ---")
critic = Agent(
role="Execution Critic",
goal="Confirm that the deployment timeline is practical, safe, and that files have been successfully structured.",
backstory="You ensure nothing breaks in production, and execution strategies contain strong fallback scripts.",
llm=local_llm
)
task = Task(
description=f"Review this implementation summary:\n\n{self.state['implementation_plan']}\n\nValidate timeline security.",
expected_output="Structured JSON confirming if execution meets criteria.",
output_json=CritiqueResult,
agent=critic
)
crew = Crew(agents=[critic], tasks=[task])
result = crew.kickoff()
if self.state["loop_counts"]["executor"] >= self.max_revisions:
return "complete"
if not result["approved"]:
print(f"❌ Rejected by Critic. Feedback: {result['feedback']}")
self.state["executor_feedback"] = result["feedback"]
self.state["loop_counts"]["executor"] += 1
return "implement_solutions"
print("✅ Implementation Strategy Approved by the Board.")
return "complete"
@listen("complete")
def complete(self):
print("\n=======================================================")
print("🎉 SUCCESS: Entire lifecycle successfully passed through the Board!")
print("Final Deployment Strategy has been executed and saved to disk.")
print("=======================================================")
# ---------------------------------------------------------------------
# Application Entry Point
# ---------------------------------------------------------------------
if __name__ == "__main__":
# Required if using SerperDevTool. Replace/mock if you use a totally local offline tool.
if "SERPER_API_KEY" not in os.environ:
os.environ["SERPER_API_KEY"] = "your_serper_api_key_here"
board_flow = BoardOfDirectorsFlow()
board_flow.kickoff()
Key Architectural Breakdown
Critique Loop Injection: By utilizing
return "current_method"inside conditional routing structures, the execution loops backward. If a Critic evaluatesapproved: false, theloop_counttracks the pass and appends the explicit criticism to the task prompt variables, forcing the agent to rewrite its solution with the updated context.Local Tool Usage: The
Opportunity ScoutandStrategic Analystagents accept theSerperDevTool(Web Search) natively, while theOperations DirectorutilizesFileWriterToolto interact with the filesystem.Local Quantization Loop Protection: Smaller 8B models can occasionally enter an infinite critique loop when failing to adhere to rigid business goals. The inclusion of a
self.max_revisionsceiling acts as a deterministic short-circuit circuit breaker to ensure your local runtime always finishes gracefully.
From <https://gemini.google.com/app/e3d8d8943d8be441> Google Gemini (3.5 Flash)
Comments
Post a Comment