How one can Design a Manufacturing-Grade CAMEL Multi-Agent System with Planning, Device Use, Self-Consistency, and Critique-Pushed Refinement

import os, sys, re, json, subprocess
from typing import Record, Dict, Any, Non-compulsory, Tuple


def _pip_install(pkgs: Record[str]):
   subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", "-U"] + pkgs)


_pip_install(["camel-ai[web_tools]~=0.2", "pydantic>=2.7", "wealthy>=13.7"])


from pydantic import BaseModel, Area
from wealthy.console import Console
from wealthy.panel import Panel
from wealthy.desk import Desk


console = Console()


def _get_colab_secret(title: str) -> Non-compulsory[str]:
   strive:
       from google.colab import userdata
       v = userdata.get(title)
       return v if v else None
   besides Exception:
       return None


def ensure_openai_key():
   if os.getenv("OPENAI_API_KEY"):
       return
   v = _get_colab_secret("OPENAI_API_KEY")
   if v:
       os.environ["OPENAI_API_KEY"] = v
       return
   strive:
       from getpass import getpass
       okay = getpass("Enter OPENAI_API_KEY (enter hidden): ").strip()
       if okay:
           os.environ["OPENAI_API_KEY"] = okay
   besides Exception:
       cross


ensure_openai_key()
if not os.getenv("OPENAI_API_KEY"):
   increase RuntimeError("OPENAI_API_KEY will not be set. Add it through Colab Secrets and techniques (OPENAI_API_KEY) or paste it when prompted.")

from camel.fashions import ModelFactory
from camel.varieties import ModelPlatformType, ModelType
from camel.brokers import ChatAgent
from camel.toolkits import SearchToolkit


def make_model(temperature: float = 0.2):
   return ModelFactory.create(
       model_platform=ModelPlatformType.OPENAI,
       model_type=ModelType.GPT_4O,
       model_config_dict={"temperature": float(temperature)},
   )


def strip_code_fences(s: str) -> str:
   s = s.strip()
   s = re.sub(r"^```(?:json)?s*", "", s, flags=re.IGNORECASE)
   s = re.sub(r"s*```$", "", s)
   return s.strip()


def extract_first_json_object(s: str) -> str:
   s2 = strip_code_fences(s)
   begin = None
   stack = []
   for i, ch in enumerate(s2):
       if ch == "{":
           if begin is None:
               begin = i
           stack.append("{")
       elif ch == "}":
           if stack:
               stack.pop()
               if not stack and begin will not be None:
                   return s2[start:i+1]
   m = re.search(r"{[sS]*}", s2)
   if m:
       return m.group(0)
   return s2

class PlanTask(BaseModel):
   id: str = Area(..., min_length=1)
   title: str = Area(..., min_length=1)
   goal: str = Area(..., min_length=1)
   deliverable: str = Area(..., min_length=1)
   tool_hints: Record[str] = Area(default_factory=checklist)
   dangers: Record[str] = Area(default_factory=checklist)


class Plan(BaseModel):
   aim: str
   assumptions: Record[str] = Area(default_factory=checklist)
   duties: Record[PlanTask]
   success_criteria: Record[str] = Area(default_factory=checklist)


class EvidenceItem(BaseModel):
   question: str
   notes: str
   key_points: Record[str] = Area(default_factory=checklist)


class Critique(BaseModel):
   score_0_to_10: float = Area(..., ge=0, le=10)
   strengths: Record[str] = Area(default_factory=checklist)
   points: Record[str] = Area(default_factory=checklist)
   fix_plan: Record[str] = Area(default_factory=checklist)


class RunConfig(BaseModel):
   aim: str
   max_tasks: int = 5
   max_searches_per_task: int = 2
   max_revision_rounds: int = 1
   self_consistency_samples: int = 2


DEFAULT_GOAL = "Create a concise, evidence-backed technical transient explaining CAMEL (the multi-agent framework), its core abstractions, and a sensible recipe to construct a tool-using multi-agent pipeline (planner/researcher/author/critic) with safeguards."


cfg = RunConfig(aim=DEFAULT_GOAL)


search_tool = SearchToolkit().search_duckduckgo

planner_system = (
   "You're a senior agent architect. Produce a compact, high-leverage plan for reaching the aim.n"
   "Return ONLY legitimate JSON that matches this schema:n"
   "{{"aim": "...", "assumptions": ["..."], "duties": "
   "[{{"id": "T1", "title": "...", "objective": "...", "deliverable": "...", "
   ""tool_hints": ["..."], "dangers": ["..."]}}], "
   ""success_criteria": ["..."]}}n"
   "Constraints: duties size <= {max_tasks}. Every activity must be executable with internet search + reasoning."
).format(max_tasks=cfg.max_tasks)


planner = ChatAgent(system_message=planner_system, mannequin=make_model(0.1))


researcher = ChatAgent(
   system_message=(
       "You're a meticulous analysis agent. Use the online search software when helpful.n"
       "You need to:n"
       "- Seek for authoritative sources (docs, official repos) first.n"
       "- Write notes which are straight related to the duty goal.n"
       "- Return ONLY legitimate JSON:n"
       "{"question": "...", "notes": "...", "key_points": ["..."]}n"
       "Don't embody markdown code fences."
   ),
   mannequin=make_model(0.2),
   instruments=[search_tool],
)


author = ChatAgent(
   system_message=(
       "You're a technical author agent. You may be given a aim, a plan, and proof notes.n"
       "Write a deliverable that's clear, actionable, and concise.n"
       "Embody:n"
       "- A crisp overviewn"
       "- Key abstractions and the way they connectn"
       "- A sensible implementation recipen"
       "- Minimal caveats/limitationsn"
       "Do NOT fabricate citations. If proof is skinny, state uncertainty.n"
       "Return plain textual content solely."
   ),
   mannequin=make_model(0.3),
)


critic = ChatAgent(
   system_message=(
       "You're a strict reviewer. Consider the draft towards the aim, correctness, and completeness.n"
       "Return ONLY legitimate JSON:n"
       "{"score_0_to_10": 0.0, "strengths": ["..."], "points": ["..."], "fix_plan": ["..."]}n"
       "Don't embody markdown code fences."
   ),
   mannequin=make_model(0.0),
)


rewriter = ChatAgent(
   system_message=(
       "You're a revising editor. Enhance the draft primarily based on critique. Protect factual accuracy.n"
       "Return the improved draft as plain textual content solely."
   ),
   mannequin=make_model(0.25),
)

def plan_goal(aim: str) -> Plan:
   resp = planner.step("GOAL:n" + aim + "nnReturn JSON plan now.")
   uncooked = resp.msgs[0].content material if hasattr(resp, "msgs") else resp.msg.content material
   js = extract_first_json_object(uncooked)
   strive:
       return Plan.model_validate_json(js)
   besides Exception:
       return Plan.model_validate(json.hundreds(js))


def research_task(activity: PlanTask, aim: str, okay: int) -> EvidenceItem:
   immediate = (
       "GOAL:n" + aim + "nnTASK:n" + activity.model_dump_json(indent=2) + "nn"
       f"Carry out analysis. Use at most {okay} internet searches. First search official documentation or GitHub if related."
   )
   resp = researcher.step(immediate)
   uncooked = resp.msgs[0].content material if hasattr(resp, "msgs") else resp.msg.content material
   js = extract_first_json_object(uncooked)
   strive:
       return EvidenceItem.model_validate_json(js)
   besides Exception:
       return EvidenceItem.model_validate(json.hundreds(js))


def draft_with_self_consistency(aim: str, plan: Plan, proof: Record[Tuple[PlanTask, EvidenceItem]], n: int) -> str:
   packed_evidence = []
   for t, ev in proof:
       packed_evidence.append({
           "task_id": t.id,
           "task_title": t.title,
           "goal": t.goal,
           "notes": ev.notes,
           "key_points": ev.key_points
       })
   payload = {
       "aim": aim,
       "assumptions": plan.assumptions,
       "duties": [t.model_dump() for t in plan.tasks],
       "proof": packed_evidence,
       "success_criteria": plan.success_criteria,
   }
   drafts = []
   for _ in vary(max(1, n)):
       resp = author.step("INPUT:n" + json.dumps(payload, ensure_ascii=False, indent=2))
       txt = resp.msgs[0].content material if hasattr(resp, "msgs") else resp.msg.content material
       drafts.append(txt.strip())
   if len(drafts) == 1:
       return drafts[0]
   chooser = ChatAgent(
       system_message=(
           "You're a selector agent. Select the very best draft amongst candidates for correctness, readability, and actionability.n"
           "Return ONLY the successful draft textual content, unchanged."
       ),
       mannequin=make_model(0.0),
   )
   resp = chooser.step("GOAL:n" + aim + "nnCANDIDATES:n" + "nn---nn".be part of([f"[DRAFT {i+1}]n{d}" for i, d in enumerate(drafts)]))
   return (resp.msgs[0].content material if hasattr(resp, "msgs") else resp.msg.content material).strip()

def critique_text(aim: str, draft: str) -> Critique:
   resp = critic.step("GOAL:n" + aim + "nnDRAFT:n" + draft + "nnReturn critique JSON now.")
   uncooked = resp.msgs[0].content material if hasattr(resp, "msgs") else resp.msg.content material
   js = extract_first_json_object(uncooked)
   strive:
       return Critique.model_validate_json(js)
   besides Exception:
       return Critique.model_validate(json.hundreds(js))


def revise(aim: str, draft: str, critique: Critique) -> str:
   resp = rewriter.step(
       "GOAL:n" + aim +
       "nnCRITIQUE:n" + critique.model_dump_json(indent=2) +
       "nnDRAFT:n" + draft +
       "nnRewrite now."
   )
   return (resp.msgs[0].content material if hasattr(resp, "msgs") else resp.msg.content material).strip()


def pretty_plan(plan: Plan):
   tab = Desk(title="Agent Plan", show_lines=True)
   tab.add_column("ID", type="daring")
   tab.add_column("Title")
   tab.add_column("Goal")
   tab.add_column("Deliverable")
   for t in plan.duties:
       tab.add_row(t.id, t.title, t.goal, t.deliverable)
   console.print(tab)


def run(cfg: RunConfig):
   console.print(Panel.match("CAMEL Superior Agentic Tutorial Runner", type="daring"))
   plan = plan_goal(cfg.aim)
   pretty_plan(plan)


   proof = []
   for activity in plan.duties[: cfg.max_tasks]:
       ev = research_task(activity, cfg.aim, cfg.max_searches_per_task)
       proof.append((activity, ev))


   console.print(Panel.match("Drafting (self-consistency)", type="daring"))
   draft = draft_with_self_consistency(cfg.aim, plan, proof, cfg.self_consistency_samples)


   for r in vary(cfg.max_revision_rounds + 1):
       crit = critique_text(cfg.aim, draft)
       console.print(Panel.match(f"Critique spherical {r+1} — rating {crit.score_0_to_10:.1f}/10", type="daring"))
       if crit.strengths:
           console.print(Panel("Strengths:n- " + "n- ".be part of(crit.strengths), title="Strengths"))
       if crit.points:
           console.print(Panel("Points:n- " + "n- ".be part of(crit.points), title="Points"))
       if crit.fix_plan:
           console.print(Panel("Repair plan:n- " + "n- ".be part of(crit.fix_plan), title="Repair plan"))
       if crit.score_0_to_10 >= 8.5 or r >= cfg.max_revision_rounds:
           break
       draft = revise(cfg.aim, draft, crit)


   console.print(Panel.match("FINAL DELIVERABLE", type="daring inexperienced"))
   console.print(draft)


run(cfg)