Multi-Agent 框架终极对比：LangGraph、CrewAI、AutoGen 谁才是真·编排之王？

bash 体验AI代码助手 代码解读复制代码pip install langgraph langchain langchain-openai

python 体验AI代码助手 代码解读复制代码import os
from typing import Annotated, TypedDict, List, Literal
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage

# 1. 定义状态 (State)
# 包含对话历史、当前代码、评审意见
class CodeReviewState(TypedDict):
    messages: List
    code: str
    review_feedback: str
    iteration_count: int

# 2. 初始化模型
# 假设使用兼容 OpenAI 协议的本地模型或云端模型
llm = ChatOpenAI(model="qwen-2.5-coder", temperature=0)

# 3. 定义节点函数

def coder_node(state: CodeReviewState):
    """Coder: 根据需求或反馈编写/修改代码"""
    messages = state["messages"]
    
    # 如果是第一次，生成代码；如果有反馈，修改代码
    if state["iteration_count"] == 0:
        prompt = "请编写一个 Python 函数，计算斐波那契数列的第 n 项。"
    else:
        prompt = f"请根据评审意见修改代码：{state['review_feedback']}\n当前代码：{state['code']}"
    
    response = llm.invoke([HumanMessage(content=prompt)] + messages)
    return {"code": response.content, "messages": state["messages"] + [response], "iteration_count": state["iteration_count"] + 1}

def reviewer_node(state: CodeReviewState):
    """Reviewer: 检查代码漏洞和规范性"""
    code = state["code"]
    prompt = f"请评审以下代码的安全性、性能和规范性。如果没有问题，回复'APPROVED'。如果有问题，列出具体问题和建议。\n代码：\n{code}"
    
    response = llm.invoke([HumanMessage(content=prompt)])
    content = response.content
    
    # 判断是否通过
    if "APPROVED" in content:
        return {"review_feedback": "APPROVED", "messages": state["messages"] + [response]}
    else:
        return {"review_feedback": content, "messages": state["messages"] + [response]}

def manager_node(state: CodeReviewState):
    """Manager: 决策是结束还是继续循环"""
    if state["review_feedback"] == "APPROVED":
        return "end"
    
    # 如果迭代超过 3 次仍未通过，强制结束防止死循环
    if state["iteration_count"] > 3:
        return "end"
    
    return "continue"

# 4. 构建图 (Graph)
workflow = StateGraph(CodeReviewState)

# 添加节点
workflow.add_node("coder", coder_node)
workflow.add_node("reviewer", reviewer_node)

# 定义边的逻辑
def route_logic(state):
    result = manager_node(state)
    if result == "end":
        return END
    return "reviewer"

# 设置边
workflow.set_entry_point("coder")
workflow.add_edge("coder", "reviewer")
workflow.add_conditional_edges(
    "reviewer",
    route_logic,
    {
        "reviewer": "coder", # 继续循环
        END: END             # 结束
    }
)

# 编译图
app = workflow.compile()

# 5. 运行测试
if __name__ == "__main__":
    initial_state = {
        "messages": [],
        "code": "",
        "review_feedback": "",
        "iteration_count": 0
    }
    
    print("开始代码评审流程...")
    final_state = app.invoke(initial_state)
    
    print("\n=== 最终代码 ===")
    print(final_state["code"])
    print("\n=== 评审结果 ===")
    print(final_state["review_feedback"])