主流 Agent 框架对比与选型

from langchain.agents import AgentExecutor, create_react_agent
from langchain.tools import Tool
from langchain_openai import ChatOpenAI
from langchain.memory import ConversationBufferMemory

# 定义工具
tools = [
    Tool(
        name="search",
        func=search_function,
        description="搜索互联网获取最新信息"
    ),
    Tool(
        name="calculator",
        func=calculate,
        description="执行精确数学运算"
    )
]

# 带记忆的 Agent
memory = ConversationBufferMemory(
    memory_key="chat_history",
    return_messages=True
)

llm = ChatOpenAI(model="gpt-4")
agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    memory=memory,
    max_iterations=10,
    handle_parsing_errors=True
)

# LangGraph：定义有向图工作流
from langgraph.graph import StateGraph, END

graph = StateGraph(AgentState)

graph.add_node("reason", reasoning_node)
graph.add_node("action", action_node)
graph.add_node("reflect", reflection_node)

graph.add_edge("reason", "action")
graph.add_conditional_edges(
    "action",
    should_continue,
    {"continue": "reflect", "end": END}
)
graph.add_edge("reflect", "reason")

app = graph.compile()

from openai import OpenAI

client = OpenAI()

# 创建助手
assistant = client.beta.assistants.create(
    name="数据分析师",
    instructions="你是数据分析专家，擅长 CSV 和 Excel 数据处理",
    model="gpt-4-turbo",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"},
        {
            "type": "function",
            "function": {
                "name": "query_database",
                "description": "查询数据库",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string"}
                    },
                    "required": ["sql"]
                }
            }
        }
    ]
)

# 创建对话线程
thread = client.beta.threads.create()

# 发送消息
client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="分析这份销售数据，按月份统计增长率"
)

# 运行助手
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id
)

# 轮询等待结果
import time
while run.status in ["queued", "in_progress", "requires_action"]:
    time.sleep(1)
    run = client.beta.threads.runs.retrieve(
        thread_id=thread.id,
        run_id=run.id
    )
    if run.status == "requires_action":
        # 处理工具调用
        for tool_call in run.required_action.submit_tool_outputs.tool_calls:
            result = execute_tool(tool_call)
            # 提交结果
            client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread.id,
                run_id=run.id,
                tool_outputs=[{
                    "tool_call_id": tool_call.id,
                    "output": json.dumps(result)
                }]
            )

# 获取最终消息
messages = client.beta.threads.messages.list(thread_id=thread.id)

import autogen

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

# 定义多个 Agent
assistant = autogen.AssistantAgent(
    name="工程师",
    llm_config=llm_config,
    system_message="你是一名资深 Python 工程师，擅长编写高质量代码。"
)

critic = autogen.AssistantAgent(
    name="评审员",
    llm_config=llm_config,
    system_message="你是一名严谨的代码评审员，找出所有潜在问题。"
)

user_proxy = autogen.UserProxyAgent(
    name="用户",
    human_input_mode="NEVER",
    code_execution_config={
        "work_dir": "coding",
        "use_docker": False
    }
)

# 启动群组对话
groupchat = autogen.GroupChat(
    agents=[user_proxy, assistant, critic],
    messages=[],
    max_round=12
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config
)

user_proxy.initiate_chat(
    manager,
    message="写一个股票数据爬虫"
)

from crewai import Agent, Task, Crew, Process

# 定义角色
researcher = Agent(
    role="市场研究员",
    goal="收集并分析市场趋势",
    backstory="经验丰富的市场分析师，擅长数据驱动决策",
    tools=[search_tool, web_scraper],
    verbose=True,
    allow_delegation=False
)

writer = Agent(
    role="内容写手",
    goal="撰写高质量的行业报告",
    backstory="资深科技编辑，擅长将复杂概念通俗化",
    verbose=True,
    allow_delegation=False
)

# 定义任务
research_task = Task(
    description="研究 2024 年 AI Agent 市场趋势，"
                "包括市场规模、主要玩家和技术方向",
    agent=researcher,
    expected_output="包含数据来源的市场分析报告"
)

write_task = Task(
    description="基于研究结果撰写 2000 字行业报告",
    agent=writer,
    expected_output="可直接发布的 Markdown 报告",
    context=[research_task]  # 依赖前置任务
)

# 组建团队
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,  # 顺序执行
    verbose=True
)

# 执行
result = crew.kickoff()