主流智能体开发框架详细解读与对比：从单 Agent 到多智能体协作的技术选型指南

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END
import operator

# 定义全局状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    current_step: str
    research_data: dict

# 定义节点函数
def researcher_node(state: AgentState) -> AgentState:
    """研究员节点：执行信息检索"""
    query = state["messages"][-1].content
    # 调用搜索工具获取数据
    data = search_tool.invoke(query)
    return {
        "research_data": data,
        "messages": [HumanMessage(content=f"检索完成：{len(data)} 条结果")]
    }

def critic_node(state: AgentState) -> AgentState:
    """批评家节点：评估检索质量"""
    data = state["research_data"]
    # LLM 评估数据质量
    evaluation = critic_llm.invoke(f"评估以下数据质量：{data}")
    return {"messages": [evaluation]}

# 条件路由函数
def should_continue(state: AgentState) -> str:
    """判断是否需要继续检索"""
    last_message = state["messages"][-1].content
    if "质量不足" in last_message:
        return "researcher"  # 循环回研究员
    return "summarizer"     # 进入总结节点

# 构建工作流
workflow = StateGraph(AgentState)
workflow.add_node("researcher", researcher_node)
workflow.add_node("critic", critic_node)
workflow.add_node("summarizer", summarizer_node)

workflow.set_entry_point("researcher")
workflow.add_edge("researcher", "critic")
workflow.add_conditional_edges("critic", should_continue)
workflow.add_edge("summarizer", END)

# 编译执行
graph = workflow.compile()
result = graph.invoke({"messages": [HumanMessage(content="调研 AI Agent 市场趋势")]})

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

# 1. 定义工具
search_tool = SerperDevTool()
scrape_tool = ScrapeWebsiteTool()

# 2. 定义 Agent（角色化）
researcher = Agent(
    role="高级市场研究员",
    goal="深入分析 {topic} 领域的市场趋势和竞争格局",
    backstory="""你是拥有 10 年经验的科技行业分析师，
    擅长从海量信息中提炼关键洞察。你注重数据准确性，
    所有结论都需要有可靠来源支撑。""",
    tools=[search_tool, scrape_tool],
    verbose=True,
    allow_delegation=False
)

analyst = Agent(
    role="数据分析师",
    goal="基于调研数据进行深度分析，输出结构化洞察",
    backstory="""你是擅长数据建模和趋势预测的分析专家，
    能够将原始数据转化为可执行的商业洞察。""",
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role="技术内容总监",
    goal="将分析结果转化为专业的技术报告",
    backstory="""你是资深技术作家，擅长将复杂的技术概念
    转化为清晰、有说服力的内容。""",
    verbose=True,
    allow_delegation=False
)

# 3. 定义任务（带依赖关系）
research_task = Task(
    description="""调研 {topic} 领域的市场现状，包括：
    1. 主要玩家和市场份额
    2. 技术发展趋势
    3. 用户需求变化
    请确保所有数据都有可靠来源引用。""",
    agent=researcher,
    expected_output="结构化的市场调研报告，包含数据来源标注",
    output_file="research_report.md"
)

analysis_task = Task(
    description="""基于调研结果，进行深度分析：
    1. SWOT 分析
    2. 机会矩阵
    3. 未来 12 个月趋势预测""",
    agent=analyst,
    expected_output="深度分析报告，包含图表和数据支撑",
    context=[research_task]  # 依赖前置任务
)

writing_task = Task(
    description="将分析结果整合为一篇 3000 字的专业技术报告",
    agent=writer,
    expected_output="完整的 Markdown 格式技术报告",
    context=[research_task, analysis_task]
)

# 4. 组建团队并执行
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, writing_task],
    process=Process.sequential,  # 顺序执行
    verbose=2,
    memory=True  # 启用团队记忆
)

# 启动执行
result = crew.kickoff(inputs={"topic": "AI Agent 开发框架"})

import autogen

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

# 定义不同角色的 Agent
coder = autogen.AssistantAgent(
    name="Coder",
    llm_config={"config_list": config_list},
    system_message="""你是资深 Python 开发工程师，
    擅长编写高质量、可维护的代码。请编写完整的可执行代码。"""
)

tester = autogen.AssistantAgent(
    name="Tester",
    llm_config={"config_list": config_list},
    system_message="""你是软件测试工程师，负责编写单元测试
    并验证代码正确性。发现问题时请明确指出。"""
)

reviewer = autogen.AssistantAgent(
    name="Reviewer",
    llm_config={"config_list": config_list},
    system_message="""你是代码审查专家，负责评估代码质量、
    安全性和可维护性。请给出评分和改进建议。"""
)

# 用户代理 Agent（执行代码）
user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    human_input_mode="TERMINATE",
    max_consecutive_auto_reply=10,
    code_execution_config={"work_dir": "coding", "use_docker": False}
)

# 创建群聊
groupchat = autogen.GroupChat(
    agents=[user_proxy, coder, tester, reviewer],
    messages=[],
    max_round=20,
    speaker_selection_method="round_robin"
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config={"config_list": config_list}
)

# 启动任务
user_proxy.initiate_chat(
    manager,
    message="""请开发一个 Python 计算器工具，
    支持加减乘除四则运算，要求有完整的单元测试。"""
)

from llama_index.core.workflow import (
    Workflow, StartEvent, StopEvent, step, Event
)
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.tools import FunctionTool

# 定义事件
class ResearchEvent(Event):
    query: str
    results: list

class AnalysisEvent(Event):
    data: list
    insights: str

# 定义工作流
class ResearchWorkflow(Workflow):
    @step()
    async def research(self, ev: StartEvent) -> ResearchEvent:
        """检索节点：执行多源检索"""
        query = ev.get("query")
        
        # 执行混合检索
        results = vector_index.retrieve(query, top_k=5)
        keyword_results = keyword_index.retrieve(query, top_k=3)
        
        # 结果去重合并
        merged = self._merge_results(results, keyword_results)
        return ResearchEvent(query=query, results=merged)
    
    @step()
    async def analyze(self, ev: ResearchEvent) -> AnalysisEvent:
        """分析节点：提炼洞察"""
        insights = await analysis_agent.achat(
            f"基于以下资料分析关键洞察：{ev.results}"
        )
        return AnalysisEvent(data=ev.results, insights=str(insights))
    
    @step()
    async def synthesize(self, ev: AnalysisEvent) -> StopEvent:
        """综合节点：生成最终报告"""
        report = await writer_agent.achat(
            f"基于数据 {ev.data} 和洞察 {ev.insights} 生成报告"
        )
        return StopEvent(result=str(report))

# 执行工作流
workflow = ResearchWorkflow(timeout=120, verbose=True)
result = await workflow.run(query="AI Agent 市场趋势")

from agents import Agent, Runner, handoff

# 定义专业 Agent
billing_agent = Agent(
    name="账单客服",
    instructions="""你是账单问题专家，负责处理账单查询、
    发票申请、退款申请等问题。""",
    tools=[get_bill_tool, apply_refund_tool]
)

technical_agent = Agent(
    name="技术支持",
    instructions="""你是技术支持专家，负责排查技术问题、
    提供解决方案。""",
    tools=[check_status_tool, restart_service_tool]
)

# 定义路由 Agent
triage_agent = Agent(
    name="客服路由",
    instructions="""根据用户问题，判断属于账单问题还是技术问题，
    并转交给对应的专业 Agent。""",
    handoffs=[billing_agent, technical_agent]
)

# 执行对话
result = Runner.run_streamed(
    triage_agent,
    "我的账单有问题，上个月多扣了钱"
)

async for event in result.stream_events():
    if event.type == "agent_response":
        print(f"[{event.agent.name}]: {event.response.output_text}")