LangGraph vs CrewAI 对比

# LangGraph —— 带检查点、需人工审批的简单智能体流程
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
from typing import TypedDict

class State(TypedDict):
    input: str
    approved: bool
    result: str

def draft_step(state: State) -> dict:
    return {"result": f"draft for: {state['input']}"}

def approval_step(state: State) -> dict:
    decision = interrupt({"question": "Approve this draft?", "draft": state["result"]})
    return {"approved": decision == "yes"}

graph = StateGraph(State)
graph.add_node("draft", draft_step)
graph.add_node("approval", approval_step)
graph.add_edge(START, "draft")
graph.add_edge("draft", "approval")
graph.add_edge("approval", END)

app = graph.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "run-1"}}
app.invoke({"input": "weekly report"}, config=config)
# 人工审批后从中断处继续执行：
app.invoke(Command(resume="yes"), config=config)

# CrewAI —— Flow + Crew，通过 @human_feedback 实现人工审批流程（crewai >= 1.8.0）
from crewai import Agent, Crew, Task
from crewai.flow.flow import Flow, listen, start
from crewai.flow.human_feedback import HumanFeedbackResult, human_feedback

writer = Agent(
    role="Content Writer",
    goal="Draft a short weekly report",
    backstory="You are a concise technical writer.",
)

draft_task = Task(
    description="Write a one-paragraph weekly report about {topic}",
    expected_output="A one-paragraph report",
    agent=writer,
)

class ReportFlow(Flow):
    @start()
    def create_draft(self):
        crew = Crew(agents=[writer], tasks=[draft_task])
        return str(crew.kickoff(inputs={"topic": "deployment metrics"}))

    # emit + llm：将自由文本反馈归约为单一结果
    @listen(create_draft)
    @human_feedback(
        message="Approve this draft, or say what needs changing:",
        emit=["approved", "rejected"],
        llm="gpt-4o-mini",
        default_outcome="rejected",
    )
    def review_draft(self, draft):
        return draft

    # HumanFeedbackResult 会传递给监听者，而不是被装饰的方法本身
    @listen("approved")
    def publish(self, result: HumanFeedbackResult):
        print(f"Published: {result.output}")

    @listen("rejected")
    def discard(self, result: HumanFeedbackResult):
        print(f"Rejected: {result.feedback}")

flow = ReportFlow()
flow.kickoff()