从 LangGraph 迁移到 Swarms：一份实操指南

{
  "job_id": "graph-workflow-abc123xyz",
  "name": "Research-Analysis-Workflow",
  "description": "A simple sequential workflow for research and analysis",
  "status": "success",
  "outputs": {
    "ResearchAgent": "Research findings on AI trends...",
    "AnalysisAgent": "Analysis of research findings..."
  },
  "usage": {
    "input_tokens": 1250,
    "output_tokens": 3200,
    "total_tokens": 4450,
    "total_cost": 0.087325,
    "cost_per_agent": 0.02
  },
  "timestamp": "2024-01-15T10:30:45.123456+00:00"
}

from swarms import Agent, GraphWorkflow

wf = GraphWorkflow(auto_compile=True)
for name in ["research", "summarize", "critique", "editor"]:
    wf.add_node(Agent(agent_name=name, model_name="gpt-4.1", max_loops=1))

wf.add_edge("research", "summarize")
wf.add_edge("research", "critique")
wf.add_edge("summarize", "editor")
wf.add_edge("critique", "editor")

wf.set_entry_points(["research"])
wf.set_end_points(["editor"])

results = wf.run(task="Assess the market for solid-state batteries.")
print(results["editor"])

import os

import httpx

BASE_URL = "https://api.swarms.world"
HEADERS = {
    "x-api-key": os.environ["SWARMS_API_KEY"],
    "Content-Type": "application/json",
}


def agent(name, prompt, model="claude-sonnet-5", **extra):
    """One graph node."""
    return {
        "agent_name": name,
        "system_prompt": prompt,
        "model_name": model,
        "max_loops": 1,
        **extra,
    }


def edge(source, target, **metadata):
    """One directed edge, with optional metadata tags."""
    e = {"source": source, "target": target}
    if metadata:
        e["metadata"] = metadata
    return e


def run_graph(name, task, agents, edges, entry_points, end_points):
    """POST /v1/graph-workflow/completions and return the parsed JSON."""
    response = httpx.post(
        f"{BASE_URL}/v1/graph-workflow/completions",
        headers=HEADERS,
        json={
            "name": name,
            "task": task,
            "agents": agents,
            "edges": edges,
            "entry_points": entry_points,
            "end_points": end_points,
            "max_loops": 1,
            "auto_compile": True,
        },
        timeout=900.0,
    )
    response.raise_for_status()
    return response.json()

from operator import add
from typing import Annotated, Literal

from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel
from typing_extensions import TypedDict

llm = ChatOpenAI(model="gpt-4.1")

class ChainState(TypedDict):
    task: str
    research: str
    analysis: str
    report: str


def research(state: ChainState):
    return {"research": llm.invoke(f"Research: {state['task']}").content}


def analyze(state: ChainState):
    return {"analysis": llm.invoke(f"Analyze:\n{state['research']}").content}


def write(state: ChainState):
    return {"report": llm.invoke(f"Write a brief from:\n{state['analysis']}").content}


builder = StateGraph(ChainState)
builder.add_node("research", research)
builder.add_node("analyze", analyze)
builder.add_node("write", write)
builder.add_edge(START, "research")
builder.add_edge("research", "analyze")
builder.add_edge("analyze", "write")
builder.add_edge("write", END)

graph = builder.compile()
print(graph.invoke({"task": "Assess the EV battery market"})["report"])

result = run_graph(
    "Research pipeline",
    "Assess the EV battery market",
    agents=[
        agent("Researcher", "Research the topic thoroughly and cite your sources."),
        agent("Analyst", "Analyze the research you receive and draw conclusions."),
        agent("Writer", "Write a one-page brief from the analysis you receive."),
    ],
    edges=[edge("Researcher", "Analyst"), edge("Analyst", "Writer")],
    entry_points=["Researcher"],
    end_points=["Writer"],
)
print(result["outputs"]["Writer"])

class LocaleState(TypedDict):
    brief: str
    copy: str
    fr: str
    de: str
    ja: str


def copywriter(state: LocaleState):
    return {"copy": llm.invoke(f"Write launch copy for: {state['brief']}").content}


def translator(key: str, language: str):
    def node(state: LocaleState):
        return {key: llm.invoke(f"Translate into {language}:\n{state['copy']}").content}
    return node


builder = StateGraph(LocaleState)
builder.add_node("copywriter", copywriter)
builder.add_edge(START, "copywriter")
for key, language in [("fr", "French"), ("de", "German"), ("ja", "Japanese")]:
    builder.add_node(key, translator(key, language))
    builder.add_edge("copywriter", key)
    builder.add_edge(key, END)

graph = builder.compile()
out = graph.invoke({"brief": "A home battery that pays for itself in six years"})
print(out["fr"], out["de"], out["ja"])

languages = ["French", "German", "Japanese"]
translators = [
    agent(f"Translator-{lang}", f"Translate the copy you receive into {lang}. Keep the tone.", model="gpt-4.1-mini")
    for lang in languages
]

result = run_graph(
    "Localization fan-out",
    "Write launch copy for a home battery that pays for itself in six years.",
    agents=[agent("Copywriter", "Write short, concrete launch copy for the product.")] + translators,
    edges=[edge("Copywriter", t["agent_name"]) for t in translators],
    entry_points=["Copywriter"],
    end_points=[t["agent_name"] for t in translators],
)
for t in translators:
    print(result["outputs"][t["agent_name"]])

class FanInState(TypedDict):
    topic: str
    notes: Annotated[list[str], add]
    report: str


def researcher(focus: str):
    def node(state: FanInState):
        return {"notes": [llm.invoke(f"Research {focus} for {state['topic']}").content]}
    return node


def synthesize(state: FanInState):
    return {"report": llm.invoke("Synthesize:\n\n" + "\n\n".join(state["notes"])).content}


builder = StateGraph(FanInState)
builder.add_node("synthesize", synthesize)
sources = {"market": "market trends", "competitors": "competitor strategies", "technology": "emerging technology"}
for name, focus in sources.items():
    builder.add_node(name, researcher(focus))
    builder.add_edge(START, name)
builder.add_edge(list(sources), "synthesize")
builder.add_edge("synthesize", END)

graph = builder.compile()
print(graph.invoke({"topic": "AI-powered SaaS"})["report"])

researchers = [
    agent("MarketResearcher", "Analyze market trends and identify opportunities."),
    agent("CompetitorAnalyst", "Analyze competitor strategies and market positioning."),
    agent("TechnologyScout", "Identify emerging technologies and innovations."),
]

result = run_graph(
    "Parallel research synthesis",
    "Strategic analysis of the AI-powered SaaS market",
    agents=researchers + [agent("StrategicSynthesizer", "Combine the research streams you receive into strategic insights.")],
    edges=[edge(r["agent_name"], "StrategicSynthesizer") for r in researchers],
    entry_points=[r["agent_name"] for r in researchers],
    end_points=["StrategicSynthesizer"],
)
print(result["outputs"]["StrategicSynthesizer"])

class DiamondState(TypedDict):
    topic: str
    research: str
    reviews: Annotated[list[str], add]
    final: str


def research(state: DiamondState):
    return {"research": llm.invoke(f"Research {state['topic']}").content}


def summarize(state: DiamondState):
    return {"reviews": [llm.invoke(f"Summarize:\n{state['research']}").content]}


def critique(state: DiamondState):
    return {"reviews": [llm.invoke(f"Critique:\n{state['research']}").content]}


def editor(state: DiamondState):
    return {"final": llm.invoke("Write the final piece from:\n\n" + "\n\n".join(state["reviews"])).content}


builder = StateGraph(DiamondState)
for name, fn in [("research", research), ("summarize", summarize), ("critique", critique), ("editor", editor)]:
    builder.add_node(name, fn)
builder.add_edge(START, "research")
builder.add_edge("research", "summarize")
builder.add_edge("research", "critique")
builder.add_edge(["summarize", "critique"], "editor")
builder.add_edge("editor", END)

graph = builder.compile()
print(graph.invoke({"topic": "solid-state batteries"})["final"])

result = run_graph(
    "Diamond",
    "Assess the market for solid-state batteries.",
    agents=[
        agent("Research", "Research the topic and list the key facts with sources."),
        agent("Summarize", "Summarize the research you receive in ten bullets."),
        agent("Critique", "Find gaps, weak sources, and missing counterarguments in the research."),
        agent("Editor", "Write the final piece from the summary and the critique you receive."),
    ],
    edges=[
        edge("Research", "Summarize"),
        edge("Research", "Critique"),
        edge("Summarize", "Editor"),
        edge("Critique", "Editor"),
    ],
    entry_points=["Research"],
    end_points=["Editor"],
)
print(result["outputs"]["Editor"])

class MeshState(TypedDict):
    topic: str
    data: Annotated[list[str], add]
    analyses: Annotated[list[str], add]
    validations: Annotated[list[str], add]
    report: str


def collector(i: int):
    def node(state: MeshState):
        return {"data": [llm.invoke(f"Collect data on {state['topic']} from source {i}").content]}
    return node


def analyst(i: int):
    def node(state: MeshState):
        return {"analyses": [llm.invoke(f"Analysis {i} of:\n" + "\n".join(state["data"])).content]}
    return node


def validator(i: int):
    def node(state: MeshState):
        return {"validations": [llm.invoke(f"Validation {i} of:\n" + "\n".join(state["analyses"])).content]}
    return node


def synthesis(state: MeshState):
    return {"report": llm.invoke("Final report from:\n" + "\n".join(state["validations"])).content}


builder = StateGraph(MeshState)
collectors = [f"collector_{i}" for i in range(1, 4)]
analysts = [f"analyst_{i}" for i in range(1, 4)]
validators = [f"validator_{i}" for i in range(1, 3)]
for i, name in enumerate(collectors, 1):
    builder.add_node(name, collector(i))
    builder.add_edge(START, name)
for i, name in enumerate(analysts, 1):
    builder.add_node(name, analyst(i))
for i, name in enumerate(validators, 1):
    builder.add_node(name, validator(i))
builder.add_node("synthesis", synthesis)
for a in analysts:
    builder.add_edge(collectors, a)
for v in validators:
    builder.add_edge(analysts, v)
builder.add_edge(validators, "synthesis")
builder.add_edge("synthesis", END)

graph = builder.compile()
print(graph.invoke({"topic": "renewable energy markets"})["report"])

collectors = [agent(f"DataCollector{i}", f"Gather data on the topic from source {i}.", model="gpt-4.1-mini") for i in range(1, 4)]
analysts = [agent(f"Analyst{i}", "Analyze the data you receive and extract key insights.") for i in range(1, 4)]
validators = [agent(f"Validator{i}", "Check the analyses you receive for accuracy and completeness.") for i in range(1, 3)]
synthesis = agent("SynthesisAgent", "Combine the validated analyses you receive into a final report.")


def all_to_all(sources, targets):
    return [edge(s["agent_name"], t["agent_name"]) for s in sources for t in targets]


result = run_graph(
    "Layered mesh",
    "Research renewable energy markets: collect, analyze, validate, synthesize.",
    agents=collectors + analysts + validators + [synthesis],
    edges=all_to_all(collectors, analysts) + all_to_all(analysts, validators) + all_to_all(validators, [synthesis]),
    entry_points=[c["agent_name"] for c in collectors],
    end_points=["SynthesisAgent"],
)
print(result["outputs"]["SynthesisAgent"])
print(result["usage"]["cost_per_agent"])  # 9 agents x $0.01

class DealState(TypedDict):
    deal: str
    legal: str
    financial: str
    technical: str
    summary: str
    risks: str


def review(key: str, focus: str):
    def node(state: DealState):
        return {key: llm.invoke(f"{focus} review of this deal: {state['deal']}").content}
    return node


def summary(state: DealState):
    notes = f"{state['legal']}\n\n{state['financial']}\n\n{state['technical']}"
    return {"summary": llm.invoke(f"Executive summary:\n{notes}").content}


def risks(state: DealState):
    return {"risks": llm.invoke(f"Risk register:\n{state['legal']}\n\n{state['financial']}").content}


builder = StateGraph(DealState)
for key, focus in [("legal", "Legal"), ("financial", "Financial"), ("technical", "Technical")]:
    builder.add_node(key, review(key, focus))
    builder.add_edge(START, key)
builder.add_node("summary", summary)
builder.add_node("risks", risks)
builder.add_edge(["legal", "financial", "technical"], "summary")
builder.add_edge(["legal", "financial"], "risks")
builder.add_edge("summary", END)
builder.add_edge("risks", END)

graph = builder.compile()
out = graph.invoke({"deal": "Acquisition of a 40-person robotics startup"})
print(out["summary"], out["risks"])

result = run_graph(
    "Deal review",
    "Acquisition of a 40-person robotics startup",
    agents=[
        agent("Legal", "Review the deal for legal exposure."),
        agent("Financial", "Review the deal's valuation and financial risk.", model="gpt-4.1"),
        agent("Technical", "Review the target's technology and team.", model="gpt-4.1"),
        agent("ExecSummary", "Write a one-page executive summary from the reviews you receive."),
        agent("RiskRegister", "Build a risk register table from the reviews you receive."),
    ],
    edges=[
        edge("Legal", "ExecSummary"),
        edge("Financial", "ExecSummary"),
        edge("Technical", "ExecSummary"),
        edge("Legal", "RiskRegister"),
        edge("Financial", "RiskRegister"),
    ],
    entry_points=["Legal", "Financial", "Technical"],
    end_points=["ExecSummary", "RiskRegister"],
)
print(result["outputs"]["ExecSummary"])
print(result["outputs"]["RiskRegister"])

from langgraph.types import Send


class MapState(TypedDict):
    companies: list[str]
    memos: Annotated[list[str], add]
    ranking: str


class MemoState(TypedDict):
    company: str


def fan_out(state: MapState):
    return [Send("write_memo", {"company": c}) for c in state["companies"]]


def write_memo(state: MemoState):
    memo = llm.invoke(f"One-paragraph investment memo on {state['company']}.").content
    return {"memos": [memo]}


def rank(state: MapState):
    return {"ranking": llm.invoke("Rank these companies:\n\n" + "\n\n".join(state["memos"])).content}


builder = StateGraph(MapState)
builder.add_node("write_memo", write_memo)
builder.add_node("rank", rank)
builder.add_conditional_edges(START, fan_out, ["write_memo"])
builder.add_edge("write_memo", "rank")
builder.add_edge("rank", END)

graph = builder.compile()
companies = ["CATL", "QuantumScape", "Solid Power", "Northvolt"]
print(graph.invoke({"companies": companies})["ranking"])

companies = ["CATL", "QuantumScape", "Solid Power", "Northvolt"]

mappers = [
    agent(
        f"Memo-{i}",
        f"Write a one-paragraph investment memo on {company}: moat, risks, recent news.",
        model="gpt-4.1-mini",
    )
    for i, company in enumerate(companies)
]
ranker = agent("Ranker", "You receive one memo per company. Rank the companies in a table and justify the order.")

result = run_graph(
    "Memo map-reduce",
    "Evaluate these battery companies as long-term investments.",
    agents=mappers + [ranker],
    edges=[edge(m["agent_name"], "Ranker") for m in mappers],
    entry_points=[m["agent_name"] for m in mappers],
    end_points=["Ranker"],
)
print(result["outputs"]["Ranker"])

from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

jurors = {
    "gpt": ChatOpenAI(model="gpt-4.1"),
    "claude": ChatAnthropic(model="claude-sonnet-5").with_fallbacks([ChatOpenAI(model="gpt-4.1")]),
    "gemini": ChatGoogleGenerativeAI(model="gemini-3.8-flash"),
}


class JuryState(TypedDict):
    question: str
    verdicts: Annotated[list[str], add]
    ruling: str


def make_juror(name: str, model):
    def juror(state: JuryState):
        answer = model.invoke(f"Answer and justify: {state['question']}").content
        return {"verdicts": [f"{name}: {answer}"]}
    return juror


def judge(state: JuryState):
    return {"ruling": llm.invoke("Weigh these answers and rule:\n\n" + "\n\n".join(state["verdicts"])).content}


builder = StateGraph(JuryState)
builder.add_node("judge", judge)
for name, model in jurors.items():
    builder.add_node(name, make_juror(name, model))
    builder.add_edge(START, name)
builder.add_edge(list(jurors), "judge")
builder.add_edge("judge", END)

graph = builder.compile()
print(graph.invoke({"question": "Is this clause enforceable in California?"})["ruling"])

juror_prompt = "Answer the question and justify your answer. End with a one-line verdict."
jurors = [
    agent("Juror-GPT", juror_prompt, model="gpt-4.1", fallback_models=["gpt-4.1-mini"]),
    agent("Juror-Claude", juror_prompt, model="claude-sonnet-5", fallback_models=["gpt-4.1"]),
    agent("Juror-Gemini", juror_prompt, model="gemini-3.8-flash", fallback_models=["claude-sonnet-5"]),
]
judge = agent(
    "Judge",
    "You receive three independent answers. Note where they agree and disagree, then give a final ruling.",
    model="claude-opus-5",
)

result = run_graph(
    "Jury",
    "Is a 24-month non-compete clause enforceable for a California employee?",
    agents=jurors + [judge],
    edges=[edge(j["agent_name"], "Judge") for j in jurors],
    entry_points=[j["agent_name"] for j in jurors],
    end_points=["Judge"],
)
print(result["outputs"]["Judge"])

class DebateState(TypedDict):
    motion: str
    pro: str
    con: str
    rebuttals: Annotated[list[str], add]
    verdict: str


def pro(state: DebateState):
    return {"pro": llm.invoke(f"Argue for: {state['motion']}").content}


def con(state: DebateState):
    return {"con": llm.invoke(f"Argue against: {state['motion']}").content}


def pro_rebuttal(state: DebateState):
    return {"rebuttals": ["PRO: " + llm.invoke(f"Rebut this, arguing for the motion:\n{state['con']}").content]}


def con_rebuttal(state: DebateState):
    return {"rebuttals": ["CON: " + llm.invoke(f"Rebut this, arguing against the motion:\n{state['pro']}").content]}


def judge(state: DebateState):
    return {"verdict": llm.invoke("Decide which side won:\n\n" + "\n\n".join(state["rebuttals"])).content}


builder = StateGraph(DebateState)
for name, fn in [("pro", pro), ("con", con), ("pro_rebuttal", pro_rebuttal), ("con_rebuttal", con_rebuttal), ("judge", judge)]:
    builder.add_node(name, fn)
builder.add_edge(START, "pro")
builder.add_edge(START, "con")
builder.add_edge("con", "pro_rebuttal")
builder.add_edge("pro", "con_rebuttal")
builder.add_edge(["pro_rebuttal", "con_rebuttal"], "judge")
builder.add_edge("judge", END)

graph = builder.compile()
print(graph.invoke({"motion": "Cities should ban gas-powered leaf blowers"})["verdict"])

result = run_graph(
    "Debate",
    "Motion: cities should ban gas-powered leaf blowers.",
    agents=[
        agent("Pro", "Argue for the motion in five strong points.", model="gpt-4.1"),
        agent("Con", "Argue against the motion in five strong points.", model="claude-sonnet-5"),
        agent("ProRebuttal", "You argue for the motion. Rebut the opposing argument you receive, point by point.", model="gpt-4.1"),
        agent("ConRebuttal", "You argue against the motion. Rebut the opposing argument you receive, point by point.", model="claude-sonnet-5"),
        agent("Judge", "You receive a rebuttal from each side of a debate. Decide which side argued better and explain why.", model="gemini-3.8-flash"),
    ],
    edges=[
        edge("Con", "ProRebuttal"),
        edge("Pro", "ConRebuttal"),
        edge("ProRebuttal", "Judge"),
        edge("ConRebuttal", "Judge"),
    ],
    entry_points=["Pro", "Con"],
    end_points=["Judge"],
)
print(result["outputs"]["Judge"])

from langgraph.types import Command


class TeamState(TypedDict):
    task: str
    notes: Annotated[list[str], add]


class Next(BaseModel):
    next: Literal["researcher", "analyst", "FINISH"]


def supervisor(state: TeamState) -> Command[Literal["researcher", "analyst", "__end__"]]:
    decision = llm.with_structured_output(Next).invoke(
        f"Task: {state['task']}\nWork so far: {state['notes']}\nWho acts next, or FINISH?"
    )
    return Command(goto=END if decision.next == "FINISH" else decision.next)


def researcher(state: TeamState) -> Command[Literal["supervisor"]]:
    note = llm.invoke(f"Research for: {state['task']}").content
    return Command(update={"notes": [note]}, goto="supervisor")


def analyst(state: TeamState) -> Command[Literal["supervisor"]]:
    note = llm.invoke(f"Analyze: {state['notes']}").content
    return Command(update={"notes": [note]}, goto="supervisor")


builder = StateGraph(TeamState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.add_edge(START, "supervisor")

graph = builder.compile()
result = graph.invoke({"task": "Competitive analysis of the AI chip market"}, config={"recursion_limit": 12})
print(result["notes"][-1])

workers = [
    agent("MarketWorker", "You receive a research plan. Carry out only its market section."),
    agent("TechWorker", "You receive a research plan. Carry out only its technology section."),
    agent("FinanceWorker", "You receive a research plan. Carry out only its financial section."),
]

result = run_graph(
    "Planner and workers",
    "Competitive analysis of the AI chip market",
    agents=[agent("Planner", "Write a research plan with market, technology, and financial sections.", model="claude-opus-5")]
    + workers
    + [agent("Reviewer", "Check the three sections you receive against the plan, fix gaps, and write the final analysis.")],
    edges=[edge("Planner", w["agent_name"]) for w in workers]
    + [edge(w["agent_name"], "Reviewer") for w in workers]
    + [edge("Planner", "Reviewer")],
    entry_points=["Planner"],
    end_points=["Reviewer"],
)
print(result["outputs"]["Reviewer"])

class LoopState(TypedDict):
    product: str
    draft: str
    feedback: str
    grade: str


class Feedback(BaseModel):
    grade: Literal["pass", "revise"]
    feedback: str


def generate(state: LoopState):
    prompt = f"Write a 120-word product description for {state['product']}."
    if state.get("feedback"):
        prompt += f"\nRevise using this feedback: {state['feedback']}"
    return {"draft": llm.invoke(prompt).content}


def evaluate(state: LoopState):
    result = llm.with_structured_output(Feedback).invoke(f"Grade this draft:\n{state['draft']}")
    return {"grade": result.grade, "feedback": result.feedback}


builder = StateGraph(LoopState)
builder.add_node("generate", generate)
builder.add_node("evaluate", evaluate)
builder.add_edge(START, "generate")
builder.add_edge("generate", "evaluate")
builder.add_conditional_edges(
    "evaluate",
    lambda state: END if state["grade"] == "pass" else "generate",
    ["generate", END],
)

graph = builder.compile()
print(graph.invoke({"product": "a home battery"}, config={"recursion_limit": 10})["draft"])

result = run_graph(
    "Unrolled reflection",
    "Write a 120-word product description for a home battery.",
    agents=[
        agent("Draft", "Write the requested copy.", model="gpt-4.1"),
        agent("Critic1", "List the three biggest weaknesses of the copy you receive.", model="claude-sonnet-5"),
        agent("Revise1", "You receive a draft and a critique. Rewrite the draft to fix every weakness.", model="gpt-4.1"),
        agent("Critic2", "List any remaining weaknesses in the copy you receive. Be strict.", model="claude-sonnet-5"),
        agent("Final", "You receive copy and a critique. Produce the final copy with every issue fixed.", model="gpt-4.1"),
    ],
    edges=[
        edge("Draft", "Critic1"),
        edge("Draft", "Revise1"),
        edge("Critic1", "Revise1"),
        edge("Revise1", "Critic2"),
        edge("Revise1", "Final"),
        edge("Critic2", "Final"),
    ],
    entry_points=["Draft"],
    end_points=["Final"],
)
print(result["outputs"]["Final"])

class RouteState(TypedDict):
    ticket: str
    route: str
    reply: str


class Route(BaseModel):
    step: Literal["billing", "technical", "general"]


def classify(state: RouteState):
    return {"route": llm.with_structured_output(Route).invoke(f"Classify: {state['ticket']}").step}


def specialist(role: str):
    def node(state: RouteState):
        return {"reply": llm.invoke(f"As the {role} team, answer: {state['ticket']}").content}
    return node


builder = StateGraph(RouteState)
builder.add_node("classify", classify)
for role in ["billing", "technical", "general"]:
    builder.add_node(role, specialist(role))
    builder.add_edge(role, END)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", lambda state: state["route"], ["billing", "technical", "general"])

graph = builder.compile()
print(graph.invoke({"ticket": "I was charged twice this month."})["reply"])

routes = {
    "Billing": "payments, invoices, refunds, or duplicate charges",
    "Technical": "bugs, API errors, or integration problems",
    "General": "anything else",
}

result = run_graph(
    "Gated support routing",
    "I was charged twice this month.",
    agents=[
        agent(
            "Triage",
            "Restate the customer's message, then end with exactly one line: ROUTE: BILLING, ROUTE: TECHNICAL, or ROUTE: GENERAL.",
            model="gpt-4.1-mini",
        )
    ]
    + [
        agent(
            name,
            f"You handle {scope}. If the input you receive does not end with ROUTE: {name.upper()}, "
            "reply with the single word SKIPPED. Otherwise, answer the customer.",
            model="gpt-4.1-mini",
        )
        for name, scope in routes.items()
    ],
    edges=[edge("Triage", name) for name in routes],
    entry_points=["Triage"],
    end_points=list(routes),
)
answer = next(v for v in result["outputs"].values() if str(v).strip() != "SKIPPED")
print(answer)

class ResearchState(TypedDict):
    topic: str
    findings: Annotated[list[str], add]


def web(state: ResearchState):
    return {"findings": [llm.invoke(f"Web research on {state['topic']}").content]}


def filings(state: ResearchState):
    return {"findings": [llm.invoke(f"Filings research on {state['topic']}").content]}


team = StateGraph(ResearchState)
team.add_node("web", web)
team.add_node("filings", filings)
team.add_edge(START, "web")
team.add_edge(START, "filings")
team.add_edge("web", END)
team.add_edge("filings", END)
research_team = team.compile()


def write_report(state: ResearchState):
    return {"findings": [llm.invoke("Write the report:\n" + "\n\n".join(state["findings"])).content]}


parent = StateGraph(ResearchState)
parent.add_node("research_team", research_team)
parent.add_node("write", write_report)
parent.add_edge(START, "research_team")
parent.add_edge("research_team", "write")
parent.add_edge("write", END)

graph = parent.compile()
print(graph.invoke({"topic": "sodium-ion batteries"})["findings"][-1])

def research_team(prefix, focus):
    workers = [
        agent(f"{prefix}-web", f"Research {focus} using public web sources.", model="gpt-4.1-mini"),
        agent(f"{prefix}-filings", f"Research {focus} using filings and papers.", model="gpt-4.1-mini"),
    ]
    lead = agent(f"{prefix}-lead", f"Merge your team's {focus} research into one briefing.")
    return {
        "agents": workers + [lead],
        "edges": [edge(w["agent_name"], lead["agent_name"]) for w in workers],
        "entry": [w["agent_name"] for w in workers],
        "exit": lead["agent_name"],
    }


teams = [research_team("market", "the market"), research_team("tech", "the technology")]
writer = agent("Writer", "Write the final report from the team briefings you receive.")

result = run_graph(
    "Nested research",
    "Sodium-ion batteries",
    agents=[a for t in teams for a in t["agents"]] + [writer],
    edges=[e for t in teams for e in t["edges"]] + [edge(t["exit"], "Writer") for t in teams],
    entry_points=[n for t in teams for n in t["entry"]],
    end_points=["Writer"],
)
print(result["outputs"]["Writer"])

# graph: any compiled StateGraph, such as one from the patterns above
graph.invoke(
    {"ticket": "Suspicious login from a new country"},
    config={"tags": ["trust-and-safety"], "metadata": {"cost_center": "support"}},
)

edges = [
    edge("Classifier", "HumanEscalation", severity="critical", priority="p0", audit_tag="trust_and_safety", cost_center="support"),
    edge("Classifier", "ReviewQueue", severity="medium", priority="p2", cost_center="support"),
    edge("Classifier", "PassiveLogger", severity="low", priority="p4", cost_center="ops"),
]

result = run_graph(
    "Moderation",
    "Suspicious login from a new country, followed by a password change.",
    agents=[
        agent("Classifier", "Classify the event's severity as critical, medium, or low, and explain why."),
        agent("HumanEscalation", "If the classification you receive is critical, draft an escalation note. Otherwise reply SKIPPED."),
        agent("ReviewQueue", "If the classification you receive is medium, write a review ticket. Otherwise reply SKIPPED."),
        agent("PassiveLogger", "If the classification you receive is low, write a one-line log entry. Otherwise reply SKIPPED."),
    ],
    edges=edges,
    entry_points=["Classifier"],
    end_points=["HumanEscalation", "ReviewQueue", "PassiveLogger"],
)

edge_index = {(e["source"], e["target"]): e.get("metadata", {}) for e in edges}
for (source, target), meta in edge_index.items():
    print(target, meta.get("priority"), str(result["outputs"].get(target, ""))[:80])