from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow.v2 import Workflow, Step, Router, Loop

agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[...],
    instructions="...",
    response_model=SomePydanticModel,
)

workflow = Workflow(
    name="my_flow",
    steps=[
        Step(name="plan", agent=planner_agent),
        Loop(
            name="execute_and_verify",
            steps=[executor_step, verifier_step],
            end_condition=lambda outputs: outputs[-1].content.is_done,
        ),
        Step(name="synthesize", agent=synthesizer_agent),
    ],
)

workflow.run(message="...")

from pydantic import BaseModel, Field
from typing import List

class DraftCode(BaseModel):
    code: str = Field(description="Python code to solve the request.")
    explanation: str = Field(description="Brief explanation of the code.")

class Critique(BaseModel):
    has_errors: bool
    is_efficient: bool
    suggested_improvements: List[str]
    critique_summary: str

class RefinedCode(BaseModel):
    refined_code: str
    refinement_summary: str

model = OpenAIChat(id="gpt-5-mini")

generator = Agent(
    name="generator",
    model=model,
    response_model=DraftCode,
    instructions="Write Python code and explain it briefly.",
)

critic = Agent(
    name="critic",
    model=model,
    response_model=Critique,
    instructions="Review code for bugs, inefficiency and style issues.",
)

refiner = Agent(
    name="refiner",
    model=model,
    response_model=RefinedCode,
    instructions="Rewrite code according to the critique.",
)

def generate_step(si):
    draft = generator.run(si.message).content
    si.workflow_session_state["draft"] = draft
    return draft

def critique_step(si):
    draft = si.workflow_session_state["draft"]
    return critic.run(f"Review this code:\n```python\n{draft.code}\n```").content

def refine_step(si):
    draft = si.workflow_session_state["draft"]
    critique = si.previous_step_output.content
    return refiner.run(
        f"Original code:\n{draft.code}\n\nCritique:\n{critique.model_dump_json()}"
    ).content

reflection_wf = Workflow(
    name="reflection",
    session_state={"draft": None},
    steps=[
        Step(name="generate", executor=generate_step),
        Step(name="critique", executor=critique_step),
        Step(name="refine", executor=refine_step),
    ],
)

用户输入
-> 模型决定调用工具
-> 工具参数
-> 工具返回
-> 模型综合回答

from agno.tools.duckduckgo import DuckDuckGoTools

def get_stock_price(symbol: str) -> str:
    """Return the latest stock price for a symbol."""
    return f"The current price of {symbol.upper()} is $172.35."

tool_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[get_stock_price, DuckDuckGoTools()],
    instructions="Use tools when the question needs real-time data.",
    show_tool_calls=True,
)

tool_agent.run("What is Apple's current stock price?")

while True:
    response = model.invoke(messages)
    messages.append(response)

    if not response.tool_calls:
        break

    for call in response.tool_calls:
        result = tool_registry[call.name](**call.args)
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": result,
        })

当前问题
-> 当前推理
-> 工具调用
-> 观察结果
-> 新推理
-> 新工具调用
...

from agno.tools.duckduckgo import DuckDuckGoTools
from agno.tools.yfinance import YFinanceTools

react_agent = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[
        DuckDuckGoTools(),
        YFinanceTools(stock_price=True, company_news=True),
    ],
    instructions=[
        "You are a research assistant.",
        "Think step by step.",
        "After each tool observation, decide what is still missing.",
    ],
    reasoning=True,
    markdown=True,
    show_tool_calls=True,
)

react_agent.print_response(
    "Based on the latest news, should I be worried about AAPL next quarter?",
    stream=True,
)

from typing import List
from pydantic import BaseModel, Field

class Plan(BaseModel):
    steps: List[str] = Field(description="Ordered atomic steps.")

session_state = {
    "plan": [],
    "intermediate": [],
}

planner = Agent(
    name="planner",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=Plan,
    instructions="Break the request into ordered atomic steps.",
)

executor = Agent(
    name="executor",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[DuckDuckGoTools()],
    instructions="Answer exactly one sub-question using tools.",
)

synthesizer = Agent(
    name="synthesizer",
    model=OpenAIChat(id="gpt-5-mini"),
    instructions="Combine intermediate findings into a final answer.",
)

def plan_step(si):
    plan = planner.run(si.message).content
    si.workflow_session_state["plan"] = list(plan.steps)
    si.workflow_session_state["intermediate"] = []
    return plan

def execute_one_step(si):
    state = si.workflow_session_state
    question = state["plan"].pop(0)
    answer = executor.run(question).content
    state["intermediate"].append(f"Q: {question}\nA: {answer}")
    return answer

def synthesize_step(si):
    notes = "\n\n".join(si.workflow_session_state["intermediate"])
    return synthesizer.run(
        f"Question: {si.message}\n\nNotes:\n{notes}\n\nFinal answer:"
    ).content

def plan_empty(outputs):
    return len(planning_wf.session_state["plan"]) == 0

planning_wf = Workflow(
    name="planning",
    session_state={"plan": [], "intermediate": []},
    steps=[
        Step(name="plan", executor=plan_step),
        Loop(
            name="execute_all",
            steps=[Step(name="execute_one", executor=execute_one_step)],
            end_condition=plan_empty,
        ),
        Step(name="synthesize", executor=synthesize_step),
    ],
)

class VerificationResult(BaseModel):
    is_successful: bool
    reasoning: str

plan -> execute -> next

plan -> execute -> verify -> continue / retry / replan

def flaky_web_search(query: str) -> str:
    """An intentionally unreliable search tool."""
    if "employee count" in query.lower():
        return "Error: API endpoint is unavailable."
    return f"Mock search result for: {query}"

verifier = Agent(
    name="verifier",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=VerificationResult,
    instructions=(
        "Decide whether the observation answers the sub-question. "
        "Treat errors, empty results and irrelevant text as failures."
    ),
)

executor = Agent(
    name="executor",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[flaky_web_search],
    instructions="Answer one sub-question using tools.",
)

def execute_step(si):
    state = si.workflow_session_state
    question = state["plan"][0]
    observation = executor.run(question).content
    state["last_question"] = question
    state["last_observation"] = observation
    return observation

def verify_step(si):
    state = si.workflow_session_state
    verdict = verifier.run(
        f"Sub-question: {state['last_question']}\n"
        f"Observation:\n{state['last_observation']}"
    ).content

    if verdict.is_successful:
        state["plan"].pop(0)
        state["intermediate"].append(
            f"Q: {state['last_question']}\nA: {state['last_observation']}"
        )
        state["retries"] = 0
    else:
        state["retries"] = state.get("retries", 0) + 1

    state["last_verdict"] = verdict
    return verdict

from agno.workflow.v2 import Router

replan_step = Step(name="replan", executor=plan_step)
noop_step = Step(name="noop", executor=lambda si: "continue")

def route_after_verify(si):
    state = si.workflow_session_state

    if not state["plan"]:
        return [noop_step]

    failed = not state["last_verdict"].is_successful
    too_many_retries = state["retries"] >= 2

    if failed and too_many_retries:
        state["retries"] = 0
        return [replan_step]

    return [noop_step]

session_state = {
    "news": None,
    "technical": None,
    "financial": None,
    "final_report": None,
}

news_analyst = Agent(
    name="news_analyst",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[DuckDuckGoTools()],
    instructions="Produce a concise section on recent financial news.",
)

technical_analyst = Agent(
    name="technical_analyst",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[YFinanceTools()],
    instructions="Analyze price action and indicators.",
)

financial_analyst = Agent(
    name="financial_analyst",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[YFinanceTools(income_statements=True, key_financial_ratios=True)],
    instructions="Analyze fundamentals.",
)

report_writer = Agent(
    name="report_writer",
    model=OpenAIChat(id="gpt-5-mini"),
    instructions="Compose a final investment memo.",
)

def news_step(si):
    out = news_analyst.run(si.message).content
    si.workflow_session_state["news"] = out
    return out

def technical_step(si):
    out = technical_analyst.run(si.message).content
    si.workflow_session_state["technical"] = out
    return out

def financial_step(si):
    out = financial_analyst.run(si.message).content
    si.workflow_session_state["financial"] = out
    return out

def write_step(si):
    s = si.workflow_session_state
    return report_writer.run(
        f"News:\n{s['news']}\n\n"
        f"Technical:\n{s['technical']}\n\n"
        f"Financial:\n{s['financial']}"
    ).content

multi_agent_wf = Workflow(
    name="multi_agent",
    session_state={},
    steps=[
        Step(name="news", executor=news_step),
        Step(name="technical", executor=technical_step),
        Step(name="financial", executor=financial_step),
        Step(name="write", executor=write_step),
    ],
)

from agno.team import Team

analysts_team = Team(
    name="analysts",
    mode="coordinate",
    model=OpenAIChat(id="gpt-5-mini"),
    members=[news_analyst, technical_analyst, financial_analyst],
    instructions=[
        "Dispatch sub-tasks to the right analyst.",
        "Collect their outputs and synthesize a final memo.",
    ],
)

from typing import Optional
from pydantic import BaseModel, Field

class BlackboardState(BaseModel):
    user_request: str
    blackboard: dict
    next_agent: Optional[str] = None
    is_complete: bool = False

class ControllerDecision(BaseModel):
    next_agent: str = Field(
        description="One of: news, technical, financial, writer, FINISH"
    )
    reasoning: str

import json

controller = Agent(
    name="controller",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=ControllerDecision,
    instructions=(
        "Inspect the blackboard and decide which specialist should run next. "
        "Return FINISH if the report is ready."
    ),
)

SPECIALISTS = {
    "news": news_analyst,
    "technical": technical_analyst,
    "financial": financial_analyst,
    "writer": report_writer,
}

def controller_step(si):
    state = si.workflow_session_state
    snapshot = json.dumps(state["blackboard"], ensure_ascii=False, indent=2)

    decision = controller.run(
        f"Request: {state['user_request']}\n\nBlackboard:\n{snapshot}"
    ).content

    state["next_agent"] = decision.next_agent
    return decision

def specialist_step(si):
    state = si.workflow_session_state

    if state["next_agent"] == "FINISH":
        return "finished"

    agent = SPECIALISTS[state["next_agent"]]
    piece = agent.run(
        f"Request: {state['user_request']}\n"
        f"Blackboard:\n{json.dumps(state['blackboard'], ensure_ascii=False)}"
    ).content

    state["blackboard"][state["next_agent"]] = piece
    return piece

def blackboard_done(outputs):
    return blackboard_wf.session_state["next_agent"] == "FINISH"

blackboard_wf = Workflow(
    name="blackboard",
    session_state={
        "user_request": "",
        "blackboard": {},
        "next_agent": None,
    },
    steps=[
        Loop(
            name="blackboard_loop",
            steps=[
                Step(name="controller", executor=controller_step),
                Step(name="specialist", executor=specialist_step),
            ],
            end_condition=blackboard_done,
        )
    ],
)

generalist = Agent(
    name="generalist",
    model=OpenAIChat(id="gpt-5-mini"),
    role="Handles general Q&A.",
    instructions="Answer general knowledge questions directly.",
)

researcher = Agent(
    name="researcher",
    model=OpenAIChat(id="gpt-5-mini"),
    role="Handles research-heavy queries.",
    tools=[DuckDuckGoTools()],
    instructions="Use search for research tasks.",
)

coder = Agent(
    name="coder",
    model=OpenAIChat(id="gpt-5-mini"),
    role="Handles Python coding tasks.",
    instructions="Write, explain and debug Python code.",
)

meta = Team(
    name="meta_controller",
    mode="route",
    model=OpenAIChat(id="gpt-5-mini"),
    members=[generalist, researcher, coder],
    instructions=(
        "Choose exactly one member based on the request type. "
        "Do not solve the task yourself."
    ),
)

from agno.workflow.v2 import Parallel

class FinalRecommendation(BaseModel):
    final_recommendation: str
    confidence_score: float
    synthesis_summary: str
    identified_opportunities: List[str]
    identified_risks: List[str]

bullish = Agent(
    name="bullish",
    model=OpenAIChat(id="gpt-5-mini"),
    instructions="Argue from a growth-oriented bullish perspective.",
)

value = Agent(
    name="value",
    model=OpenAIChat(id="gpt-5-mini"),
    instructions="Analyze margin of safety and valuation risk.",
)

quant = Agent(
    name="quant",
    model=OpenAIChat(id="gpt-5-mini"),
    instructions="Focus on quantitative ratios and trends.",
)

aggregator = Agent(
    name="aggregator",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=FinalRecommendation,
    instructions=(
        "Synthesize different views. Preserve important disagreements "
        "instead of averaging them away."
    ),
)

def run_view(agent):
    def step(si):
        out = agent.run(si.message).content
        si.workflow_session_state.setdefault("views", {})[agent.name] = out
        return out
    return step

def aggregate_step(si):
    views = si.workflow_session_state["views"]
    body = "\n\n".join(f"[{name}]\n{view}" for name, view in views.items())
    return aggregator.run(f"Question: {si.message}\n\nViews:\n{body}").content

ensemble_wf = Workflow(
    name="ensemble",
    session_state={"views": {}},
    steps=[
        Parallel(
            name="parallel_views",
            steps=[
                Step(name="bullish", executor=run_view(bullish)),
                Step(name="value", executor=run_view(value)),
                Step(name="quant", executor=run_view(quant)),
            ],
        ),
        Step(name="aggregate", executor=aggregate_step),
    ],
)

from agno.memory.v2.memory import Memory
from agno.memory.v2.db.sqlite import SqliteMemoryDb
from agno.knowledge.text import TextKnowledgeBase
from agno.vectordb.lancedb import LanceDb, SearchType
from agno.embedder.openai import OpenAIEmbedder
from agno.storage.sqlite import SqliteStorage

memory = Memory(
    db=SqliteMemoryDb(
        table_name="user_memories",
        db_file="tmp/memory.db",
    ),
    model=OpenAIChat(id="gpt-5-mini"),
)

knowledge = TextKnowledgeBase(
    path="data/facts",
    vector_db=LanceDb(
        table_name="facts",
        uri="tmp/lancedb",
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

mem_agent = Agent(
    name="memorized_agent",
    model=OpenAIChat(id="gpt-5-mini"),
    memory=memory,
    enable_agentic_memory=True,
    enable_user_memories=True,
    knowledge=knowledge,
    search_knowledge=True,
    add_history_to_messages=True,
    num_history_responses=5,
    storage=SqliteStorage(
        table_name="sessions",
        db_file="tmp/sessions.db",
    ),
    markdown=True,
)

mem_agent.print_response(
    "I'm allergic to peanuts and prefer low-carb meals. Remember that.",
    user_id="alice",
)

mem_agent.print_response(
    "Suggest a dinner plan for me.",
    user_id="alice",
)

class Node(BaseModel):
    id: str = Field(description="Entity name or identifier.")
    type: str = Field(description="Entity type, e.g. Person or Company.")

class Relationship(BaseModel):
    source: Node
    target: Node
    type: str = Field(description="Relationship verb in ALL_CAPS.")
    properties: dict = Field(default_factory=dict)

class KnowledgeGraph(BaseModel):
    relationships: List[Relationship]

graph_maker = Agent(
    name="graph_maker",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=KnowledgeGraph,
    instructions=(
        "Extract entities and relationships from text. "
        "Relationship type should be an ALL_CAPS verb."
    ),
)

kg = graph_maker.run(
    "Tim Cook is the CEO of Apple. Apple acquired Beats in 2014."
).content

from agno.tools.neo4j import Neo4jTools

graph_query_agent = Agent(
    name="graph_query",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[
        Neo4jTools(
            url="bolt://localhost:7687",
            user="neo4j",
            password="password",
        )
    ],
    instructions=(
        "Answer questions over a Neo4j knowledge graph. "
        "Generate Cypher, run it, then synthesize the answer."
    ),
)

graph_query_agent.print_response(
    "Which companies did Apple acquire and in which year?"
)

class ToTState(BaseModel):
    problem: str
    active_paths: list
    solution: list | None = None

class PuzzleState(BaseModel):
    left_bank: frozenset = Field(
        default_factory=lambda: frozenset({"wolf", "goat", "cabbage"})
    )
    right_bank: frozenset = Field(default_factory=frozenset)
    boat_location: str = "left"
    move_description: str = "Initial state."

    class Config:
        arbitrary_types_allowed = True

    def is_valid(self) -> bool:
        dangerous_pairs = [("wolf", "goat"), ("goat", "cabbage")]
        unguarded_bank = (
            self.left_bank if self.boat_location == "right"
            else self.right_bank
        )
        return not any(
            {a, b}.issubset(unguarded_bank)
            for a, b in dangerous_pairs
        )

    def is_goal(self) -> bool:
        return self.right_bank == frozenset({"wolf", "goat", "cabbage"})

def expand(state: PuzzleState) -> list[PuzzleState]:
    """
    Generate valid next states.
    LLM can propose moves, but code must validate them.
    """
    ...

def solve_with_tot(initial: PuzzleState, max_depth: int = 10):
    active_paths = [[initial]]

    for _ in range(max_depth):
        new_paths = []

        for path in active_paths:
            current = path[-1]

            for next_state in expand(current):
                if next_state in path:
                    continue

                new_path = path + [next_state]

                if next_state.is_goal():
                    return new_path

                new_paths.append(new_path)

        active_paths = new_paths

    return None

import copy
import numpy as np
from pydantic import BaseModel, Field

class Portfolio(BaseModel):
    cash: float = 10000.0
    shares: int = 0

    def value(self, price: float) -> float:
        return self.cash + self.shares * price

class MarketSimulator(BaseModel):
    day: int = 0
    price: float = 100.0
    volatility: float = 0.1
    drift: float = 0.01
    portfolio: Portfolio = Field(default_factory=Portfolio)

    def step(self, action: str, amount: float = 0.0):
        if action == "buy":
            count = int(amount)
            cost = count * self.price
            if self.portfolio.cash >= cost:
                self.portfolio.shares += count
                self.portfolio.cash -= cost

        elif action == "sell":
            count = min(int(amount), self.portfolio.shares)
            self.portfolio.shares -= count
            self.portfolio.cash += count * self.price

        self.price *= 1 + float(np.random.normal(self.drift, self.volatility))
        self.day += 1

REAL = MarketSimulator()

def simulate_action(action: str, amount: float, horizon: int = 5) -> str:
    sim = copy.deepcopy(REAL)
    sim.step(action, amount)

    for _ in range(horizon - 1):
        sim.step("hold")

    return f"Simulated value after {horizon} days: ${sim.portfolio.value(sim.price):.2f}"

def execute_action(action: str, amount: float) -> str:
    REAL.step(action, amount)
    return f"Executed {action} {amount}. Portfolio value: ${REAL.portfolio.value(REAL.price):.2f}"

trader = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[simulate_action, execute_action],
    instructions=[
        "Before calling execute_action, call simulate_action.",
        "If simulation is worse than holding, do not execute.",
    ],
    show_tool_calls=True,
)

import datetime
import hashlib
from typing import List

def publish_post(content: str, hashtags: List[str], dry_run: bool = True) -> str:
    ts = datetime.datetime.now().isoformat()
    full_text = f"{content}\n\n" + " ".join(f"#{tag}" for tag in hashtags)

    if dry_run:
        return f"[DRY RUN @ {ts}] Would publish:\n---\n{full_text}\n---"

    post_id = hashlib.md5(full_text.encode()).hexdigest()[:8]
    return f"[LIVE @ {ts}] Published id={post_id}"

proposer = Agent(
    name="proposer",
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[publish_post],
    instructions=[
        "When asked to publish, call publish_post with dry_run=True.",
        "After preview, stop and wait for approval.",
        "Do not call dry_run=False by yourself.",
    ],
    show_tool_calls=True,
)

def propose_step(si):
    return proposer.run(si.message).content

def approve_step(si):
    preview = si.previous_step_output.content
    print(f"\n--- PREVIEW ---\n{preview}\n---")
    decision = input("Approve and go live? (y/n): ").strip().lower()
    si.workflow_session_state["approved"] = decision == "y"
    return decision

def commit_step(si):
    if not si.workflow_session_state.get("approved"):
        return "Rejected. Nothing was published."

    return proposer.run(
        "Publish the approved post by calling publish_post with dry_run=False."
    ).content

dry_run_wf = Workflow(
    name="dry_run",
    session_state={"approved": False},
    steps=[
        Step(name="propose", executor=propose_step),
        Step(name="approve", executor=approve_step),
        Step(name="commit", executor=commit_step),
    ],
)

from typing import Optional

class MetacognitiveAnalysis(BaseModel):
    confidence: float = Field(description="0.0 to 1.0")
    strategy: str = Field(description="reason_directly | use_tool | escalate")
    reasoning: str
    tool_to_use: Optional[str] = None

AGENT_SELF_MODEL = {
    "knowledge_domains": ["general health", "nutrition", "exercise"],
    "tools_available": ["symptom_checker"],
    "confidence_threshold": 0.7,
    "high_risk_topics": [
        "prescription dosage",
        "emergency medical advice",
    ],
}

self_model_agent = Agent(
    name="self_model",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=MetacognitiveAnalysis,
    instructions=(
        f"Your self-model is: {AGENT_SELF_MODEL}. "
        "Estimate confidence and choose one strategy: "
        "reason_directly, use_tool or escalate."
    ),
)

def symptom_checker(symptoms: str) -> str:
    return f"Reference information for: {symptoms}"

responder = Agent(
    model=OpenAIChat(id="gpt-5-mini"),
    tools=[symptom_checker],
)

def analyze_self_step(si):
    analysis = self_model_agent.run(si.message).content
    si.workflow_session_state["analysis"] = analysis
    return analysis

def route_strategy(si):
    analysis = si.workflow_session_state["analysis"]

    if analysis.strategy == "reason_directly":
        return [
            Step(
                name="answer",
                executor=lambda s: responder.run(s.message).content,
            )
        ]

    if analysis.strategy == "use_tool":
        return [
            Step(
                name="tool_answer",
                executor=lambda s: responder.run(
                    f"Use {analysis.tool_to_use} to answer: {s.message}"
                ).content,
            )
        ]

    return [
        Step(
            name="escalate",
            executor=lambda s: (
                "I am not confident or not allowed to answer this. "
                "Escalating to a human expert."
            ),
        )
    ]

metacog_wf = Workflow(
    name="metacognitive",
    session_state={},
    steps=[
        Step(name="self_model", executor=analyze_self_step),
        Router(name="route_strategy", selector=route_strategy),
    ],
)

class EmailCritique(BaseModel):
    is_approved: bool
    feedback: str

class MarketingEmail(BaseModel):
    subject: str
    body: str

generator = Agent(
    name="email_generator",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=MarketingEmail,
    instructions="Write a marketing email. Follow feedback if provided.",
)

critic = Agent(
    name="email_critic",
    model=OpenAIChat(id="gpt-5-mini"),
    response_model=EmailCritique,
    instructions=(
        "Approve only if the subject is compelling, "
        "the body has a clear CTA, and the tone is on-brand."
    ),
)

class GoldStandardMemory:
    def __init__(self):
        self.examples: list[MarketingEmail] = []

    def few_shot_block(self) -> str:
        if not self.examples:
            return "No gold examples yet."

        return "\n\n---\n\n".join(
            f"Subject: {e.subject}\nBody:\n{e.body}"
            for e in self.examples[-3:]
        )

    def add(self, email: MarketingEmail):
        self.examples.append(email)

GOLD = GoldStandardMemory()

def generate_email_step(si):
    state = si.workflow_session_state
    previous = state.get("last_email")
    critique = state.get("last_critique")

    prompt = f"Task: {si.message}\n\nGold examples:\n{GOLD.few_shot_block()}\n\n"

    if previous and critique:
        prompt += (
            f"Previous draft:\nSubject: {previous.subject}\nBody:\n{previous.body}\n\n"
            f"Feedback: {critique.feedback}\n"
            "Rewrite to address the feedback."
        )

    email = generator.run(prompt).content
    state["last_email"] = email
    state["revision"] = state.get("revision", 0) + 1
    return email

def critique_email_step(si):
    email = si.workflow_session_state["last_email"]
    verdict = critic.run(
        f"Review:\nSubject: {email.subject}\nBody:\n{email.body}"
    ).content

    si.workflow_session_state["last_critique"] = verdict

    if verdict.is_approved:
        GOLD.add(email)

    return verdict

def should_stop(outputs):
    state = self_improve_wf.session_state
    verdict = state.get("last_critique")

    if verdict and verdict.is_approved:
        return True

    return state.get("revision", 0) >= 3

self_improve_wf = Workflow(
    name="self_improve",
    session_state={"revision": 0},
    steps=[
        Loop(
            name="refine_loop",
            steps=[
                Step(name="generate", executor=generate_email_step),
                Step(name="critique", executor=critique_email_step),
            ],
            end_condition=should_stop,
        )
    ],
)

class CellAgent(BaseModel):
    type: str  # EMPTY | OBSTACLE | GOAL
    pathfinding_value: float = float("inf")

    def update(self, neighbors: list["CellAgent"]):
        if self.type == "OBSTACLE":
            return

        if self.type == "GOAL":
            self.pathfinding_value = 0
            return

        best_neighbor = min(
            (n.pathfinding_value for n in neighbors),
            default=float("inf"),
        )

        self.pathfinding_value = min(
            self.pathfinding_value,
            best_neighbor + 1,
        )

import copy

def neighbors_of(snapshot, row, col):
    height = len(snapshot)
    width = len(snapshot[0])
    result = []

    for dr, dc in [(-1, 0), (1, 0), (0, -1), (0, 1)]:
        nr, nc = row + dr, col + dc

        if 0 <= nr < height and 0 <= nc < width:
            result.append(snapshot[nr][nc])

    return result

def run_ca(grid, steps=50):
    for _ in range(steps):
        snapshot = [
            [copy.deepcopy(cell) for cell in row]
            for row in grid
        ]

        for r in range(len(grid)):
            for c in range(len(grid[0])):
                grid[r][c].update(neighbors_of(snapshot, r, c))

rule_designer = Agent(
    name="rule_designer",
    model=OpenAIChat(id="gpt-5-mini"),
    instructions=(
        "Design local update rules for a cellular automaton. "
        "Each cell can only use neighbor states."
    ),
)