AI Agent 开发框架深度横评：图范式 vs 角色模型 vs 对话协作 vs 声明编译

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    tool_results: dict
    retry_count: int

# 定义节点
def analyze_intent(state: AgentState) -> AgentState:
    """分析用户意图节点"""
    user_input = state["messages"][-1]
    # 调用 LLM 分析是否需要工具调用
    return {"messages": [{"role": "assistant", "content": "analyzing..."}]}

def execute_tool(state: AgentState) -> AgentState:
    """执行工具节点"""
    tool_name = state["messages"][-1].get("tool_call")
    result = call_tool(tool_name)
    return {"tool_results": {tool_name: result}}

def generate_response(state: AgentState) -> AgentState:
    """生成回答节点"""
    result = state["tool_results"]
    response = llm.generate(f"Based on: {result}")
    return {"messages": [{"role": "assistant", "content": response}]}

# 条件边：决定下一步走哪条路径
def should_use_tool(state: AgentState) -> str:
    if state["messages"][-1].get("tool_call"):
        return "tool"
    return "direct"

def should_retry(state: AgentState) -> str:
    if state.get("retry_count", 0) < 3:
        return "retry"
    return "fallback"

# 构建图
graph = StateGraph(AgentState)
graph.add_node("analyze", analyze_intent)
graph.add_node("tool", execute_tool)
graph.add_node("generate", generate_response)
graph.add_node("fallback", lambda s: {"messages": [{"role": "assistant", "content": "抱歉，无法处理"}]})

graph.set_entry_point("analyze")
graph.add_conditional_edges("analyze", should_use_tool, {"tool": "tool", "direct": "generate"})
graph.add_conditional_edges("tool", should_retry, {"retry": "tool", "fallback": "fallback"})
graph.add_edge("generate", END)
graph.add_edge("fallback", END)

app = graph.compile()

from crewai import Agent, Task, Crew

# 定义角色：10 行代码
researcher = Agent(
    role="高级研究员",
    goal="搜索并分析最新的技术趋势",
    backstory="你有 10 年技术研究经验，擅长从海量信息中提取关键洞察",
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="技术写手",
    goal="根据研究结果撰写深度技术文章",
    backstory="你是资深技术作家，擅长将复杂概念解释得清晰易懂",
    verbose=True,
)

editor = Agent(
    role="主编",
    goal="审核和修改文章，确保质量和准确性",
    backstory="你是技术杂志主编，有严格的编辑标准",
    verbose=True,
)

# 分配任务
research_task = Task(
    description="搜索 2026 年 AI Agent 框架的最新发展趋势",
    expected_output="一份包含 5 个关键趋势的研究简报",
    agent=researcher,
)

write_task = Task(
    description="基于研究简报撰写一篇 3000 字的技术文章",
    expected_output="一篇结构完整的技术文章",
    agent=writer,
)

edit_task = Task(
    description="审核文章，修正事实错误，优化表达",
    expected_output="一篇经过编辑的高质量文章",
    agent=editor,
)

# 创建并执行 Crew
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, write_task, edit_task],
    verbose=True,
)

result = crew.kickoff()
print(result)

import autogen

# 配置 LLM
config_list = [
    {"model": "gpt-4o", "api_key": "sk-xxx"},
]

# 定义多 Agent 对话组
user_proxy = autogen.UserProxyAgent(
    name="用户代理",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=10,  # 最多 10 轮自动对话
    code_execution_config={"work_dir": "coding"},
)

product_manager = autogen.AssistantAgent(
    name="产品经理",
    llm_config={"config_list": config_list},
    system_message="你是产品经理。你的任务是理解用户需求，定义产品规格。",
)

architect = autogen.AssistantAgent(
    name="架构师",
    llm_config={"config_list": config_list},
    system_message="你是系统架构师。你的任务是根据产品规格设计技术架构。",
)

programmer = autogen.AssistantAgent(
    name="程序员",
    llm_config={"config_list": config_list},
    system_message="你是程序员。你的任务是根据架构设计编写代码。",
)

# 启动多 Agent 对话
user_proxy.initiate_chat(
    product_manager,
    message="我想做一个 AI 客服系统，需要处理多轮对话、知识库检索和工单创建",
)

# Agent 之间会自动进行对话和协作

import dspy

# 声明：定义 LLM 和任务
llm = dspy.OpenAI(model='gpt-4o', max_tokens=1000)
dspy.settings.configure(lm=llm)

# 声明：定义 Agent 的行为规范
class CodeReviewer(dspy.Signature):
    """审查代码的质量、安全性和可读性"""
    code = dspy.InputField(desc="待审查的代码")
    issues = dspy.OutputField(desc="发现的问题列表，JSON 格式")
    suggestions = dspy.OutputField(desc="改进建议")
    score = dspy.OutputField(desc="代码质量评分，1-10")

# 声明：创建编译后的 Agent
reviewer = dspy.Predict(CodeReviewer)

# 编译：使用示例数据自动优化
trainset = [
    dspy.Example(
        code="def add(a, b): return a + b",
        issues="[]",
        suggestions="函数命名可以更具体",
        score="8"
    ).with_inputs("code"),
    # 更多示例...
]

# 编译优化
teleprompter = dspy.teleprompt.LabeledFewShot()
compiled_reviewer = teleprompter.compile(reviewer, trainset=trainset)

# 运行优化后的 Agent
result = compiled_reviewer(code="复杂代码...")
print(result.issues)
print(result.suggestions)
print(result.score)

// 构建抽象层：框架无关的 Agent 接口
interface Agent {
  run(input: string): Promise<AgentResult>;
  getCapabilities(): AgentCapabilities;
}

interface AgentResult {
  output: string;
  toolCalls: ToolCall[];
  metadata: Record<string, any>;
}

interface AgentCapabilities {
  supportsStreaming: boolean;
  supportsHumanInLoop: boolean;
  maxContextLength: number;
}

// LangGraph 适配器
class LangGraphAgent implements Agent {
  constructor(private graph: CompiledStateGraph) {}
  
  async run(input: string): Promise<AgentResult> {
    const result = await this.graph.invoke({
      messages: [{ role: "user", content: input }],
    });
    return this.toAgentResult(result);
  }
  
  getCapabilities(): AgentCapabilities {
    return {
      supportsStreaming: true,
      supportsHumanInLoop: true,
      maxContextLength: 128000,
    };
  }
  
  private toAgentResult(result: any): AgentResult {
    return {
      output: result.messages[result.messages.length - 1].content,
      toolCalls: result.toolCalls || [],
      metadata: { framework: "langgraph" },
    };
  }
}

// CrewAI 适配器
class CrewAIAgent implements Agent {
  constructor(private crew: Crew) {}
  
  async run(input: string): Promise<AgentResult> {
    const result = await this.crew.kickoff({ input });
    return {
      output: result.raw,
      toolCalls: [],
      metadata: { framework: "crewai" },
    };
  }
  
  getCapabilities(): AgentCapabilities {
    return {
      supportsStreaming: false,
      supportsHumanInLoop: false,
      maxContextLength: 32000,
    };
  }
}

// 业务代码只依赖接口，不依赖具体框架
async function processRequest(agent: Agent, input: string) {
  if (!agent.getCapabilities().supportsHumanInLoop) {
    console.warn("当前框架不支持人类介入");
  }
  return agent.run(input);
}

// 混合框架架构示例：用 MCP 桥接不同框架
interface MCPTool {
  name: string;
  execute(args: any): Promise<any>;
}

// 来自 LangGraph 的工具
const langGraphTools: MCPTool[] = [
  { name: "code_runner", execute: runCode },
  { name: "db_query", execute: queryDB },
];

// 来自 CrewAI 的工具
const crewAITools: MCPTool[] = [
  { name: "web_search", execute: searchWeb },
  { name: "file_writer", execute: writeFile },
];

// 统一工具注册表
const toolRegistry = new Map<string, MCPTool>();
[...langGraphTools, ...crewAITools].forEach(t => toolRegistry.set(t.name, t));

// 任何框架的 Agent 都可以使用这些工具
async function executeTool(name: string, args: any) {
  const tool = toolRegistry.get(name);
  if (!tool) throw new Error(`Unknown tool: ${name}`);
  return tool.execute(args);
}