AI Agent框架之争：盘点8大AI Agent开发框架的核心技术与工业级应用

from langgraph.graph import StateGraph, MessagesState  
from langchain_openai import ChatOpenAI  
# 定义持久化状态（存储对话历史）  
class AgentState(TypedDict):  
    messages: List[BaseMessage]  
# 构建工作流  
workflow = StateGraph(AgentState)  
# 添加节点：LLM调用  
def call_model(state):  
    model = ChatOpenAI(model="gpt-4o")  
    return {"messages": [model.invoke(state["messages"])]}  
workflow.add_node("assistant", call_model)  
# 条件分支：根据输出决定是否结束  
def should_continue(state):  
    last_msg = state["messages"][-1].content  
    if "[END]" in last_msg:  
        return "end"  
    return "assistant"  
workflow.add_conditional_edges("assistant", should_continue)  
workflow.set_entry_point("assistant")

from autogen import AssistantAgent, UserProxyAgent  
# 创建代理角色  
coder = AssistantAgent("coder", llm_config={"model": "gpt-4-turbo"})  
reviewer = AssistantAgent("reviewer", system_message="你是一名资深代码审查员")  
user_proxy = UserProxyAgent("user", human_input_mode="TERMINATE")  
# 定义协作流程  
def solve_bug(task):  
    user_proxy.initiate_chat(  
        coder,  
        message=f"修复这段代码的bug：\n{task}",  
        subsequent_handler=lambda msg: reviewer.send(msg, coder)  
    )  
solve_bug("def calc(a,b): return a/b")  # 触发零除错误处理流程

group_chat = GroupChat(  
    agents=[test_manager, test_generator, test_runner],  
    roles=["设计测试策略", "生成测试用例", "执行测试"]  
)  
test_manager.initiate_chat("为登录模块生成边界测试")

from crewai import Agent, Task, Crew, Process  
# 角色定义  
researcher = Agent(  
    role="市场研究员",  
    goal="发现行业趋势",  
    tools=[web_search_tool]  
)  
# 任务编排  
trend_task = Task(  
    description="分析2025年AI代理技术趋势",  
    agent=researcher,  
    expected_output="包含TOP3趋势的Markdown报告"  
)  
# 流程执行  
crew = Crew(  
    agents=[researcher],  
    tasks=[trend_task],  
    process=Process.sequential  
)  
result = crew.kickoff()

from openai_agents import Agent, tool  
@tool  
def stock_price(symbol: str) -> float:  
    """查询股票实时价格"""  
    return yf.Ticker(symbol).history().iloc[-1].Close  
# 创建Agent并绑定工具  
agent = Agent(tools=[stock_price])  
print(agent.run("苹果公司当前股价是多少？"))

agent.connect_mcp_service("financial_tools")  
agent.run("对比TSLA和AAPL的市盈率")

from metagpt.roles import ProductManager, Architect, Engineer  
pm = ProductManager()  
architect = Architect()  
engineer = Engineer()  
# 需求驱动开发  
pm.run("开发短视频推荐算法")  
architect.receive(pm.output)  # 接收PRD文档  
engineer.receive(architect.output)  # 获取设计文档并编码

from pydantic_ai import Agent  
from pydantic import BaseModel  
class MarketReport(BaseModel):  
    trends: list[str]  
    risk_factors: list[str]  
# 强制结构化输出  
agent = Agent(model="gemini-1.5", output_type=MarketReport)  
report = agent.run("生成量子计算市场报告")  
print(report.trends[0])  # 直接访问结构化字段

# AutoGen混合模型配置  
coder = AssistantAgent(llm_config={"model": "gpt-4-turbo"})  
reviewer = AssistantAgent(llm_config={"model": "claude-3-sonnet"})