AI Agent框架哪家强：Google ADK、Autogen、Langchain与CrewAI

# 需要：pip install litellmfrom google.adk.agents importAgentfrom google.adk.llm_config importLiteLlm# 假设的 LiteLLM 集成封装
openai_llm =LiteLlm(model="openai/gpt-4o-mini")# 通过 LiteLLM 指定提供商/模型
openai_agent =Agent(    name="openai_via_adk",    llm=openai_llm,# 传递 LiteLLM 配置    instruction="你由 OpenAI 提供支持，通过 ADK + LiteLLM 访问。")print("ADK 智能体已配置为通过 LiteLLM 使用 OpenAI。")# 需要设置 OPENAI_API_KEY 环境变量

from autogen importAssistantAgent, config_list_from_json
config_list_openai = config_list_from_json(env_or_file="OAI_CONFIG_LIST")assistant_openai =AssistantAgent(    name="openai_assistant",    llm_config={"config_list": config_list_openai})print("Autogen 智能体通过 OAI_CONFIG_LIST 配置完成。")
# OAI_CONFIG_LIST 是指定 API 密钥和端点的 JSON 文件或环境变量# 例如：[{"model": "gpt-4", "api_key": "..."}]

from langchain_openai importChatOpenAI# 显式导入 OpenAIfrom langchain_core.prompts importChatPromptTemplatefrom langchain_core.output_parsers importStrOutputParser
# 实例化特定提供商的模型类openai_llm =ChatOpenAI(model="gpt-4o-mini", temperature=0)prompt =ChatPromptTemplate.from_template("告诉我关于 {topic} 的信息")chain_openai = prompt | openai_llm |StrOutputParser()print("Langchain 使用显式导入的 ChatOpenAI 模型。")# result = chain_openai.invoke({"topic": "Langchain"})

from crewai importAgentfrom langchain_community.llms importOllama# 需要 langchain-community
# 概念性：通过 Ollama 配置本地模型连接local_llm =Ollama(model="llama3")# 假设 Ollama 服务器正在运行
local_agent =Agent(    role='编码者',    goal='编写 Python 代码',    backstory='...',    llm=local_llm # 传递本地大语言模型实例)print("CrewAI 智能体概念性使用传入的本地 Ollama 实例。")

from google.adk.agents importAgent,SequentialAgent
step1 =Agent(name="step1", model="gemini-2.0-flash", instruction="...")step2 =Agent(name="step2", model="gemini-2.0-flash", instruction="...")
# 使用特定的工作流程智能体类pipeline =SequentialAgent(    name="my_sequential_task",    sub_agents=[step1, step2]# 通过参数配置)print("ADK：实例化特定工作流程智能体（SequentialAgent）。")

from autogen importAssistantAgent,UserProxyAgent, config_list_from_json
config_list = config_list_from_json("OAI_CONFIG_LIST")llm_config ={"config_list": config_list}
# 用于常见角色的高层智能体类型assistant =AssistantAgent(name="assistant", llm_config=llm_config)user_proxy =UserProxyAgent(name="user_proxy", human_input_mode="NEVER", code_execution_config=False)print("Autogen：使用高层 AssistantAgent 和 UserProxyAgent。")

from autogen importConversableAgent# 提供更多控制的基类
classMyCustomAgent(ConversableAgent):# 继承以实现自定义逻辑def __init__(self, name, custom_param,**kwargs):super().__init__(name,**kwargs)self.custom_param = custom_param# 重写方法如 generate_reply 以实现自定义行为# ...# custom_agent = MyCustomAgent(...)print("Autogen：概念性自定义智能体，继承自 ConversableAgent。")

from langchain_openai importChatOpenAIfrom langchain_core.prompts importChatPromptTemplatefrom langchain.agents import create_tool_calling_agent,AgentExecutorfrom langchain_core.tools import tool
@tooldef get_length(s: str)->int:"""获取字符串长度。"""return len(s)tools =[get_length]model =ChatOpenAI(model="gpt-4o-mini")# 使用工厂函数组装智能体可运行对象# 注意：提示结构对这些智能体有特定要求prompt =ChatPromptTemplate.from_messages([("system","你是一个有用的助手。"),("human","{input}"),("placeholder","{agent_scratchpad}")# 关键占位符])agent_runnable = create_tool_calling_agent(model, tools, prompt)agent_executor =AgentExecutor(agent=agent_runnable, tools=tools, verbose=True)print("Langchain：使用工厂函数创建智能体。")# result = agent_executor.invoke({"input": "‘hello’ 的长度是多少？"})

from crewai importAgentfrom langchain_openai importChatOpenAI
llm =ChatOpenAI(model="gpt-4o-mini")
# 具有更多特定配置选项的智能体senior_researcher =Agent(    role='高级 AI 研究员',    goal='探索尖端 AI 技术',    backstory='拥有数十年 AI 研究经验的博士。',    llm=llm,    allow_delegation=False,# 禁止该智能体委托任务    max_iter=5,# 限制其任务的迭代次数    verbose=True)print("CrewAI：具有附加配置（allow_delegation、max_iter）的智能体。")

from google.adk.agents importAgent,ParallelAgent
task_a =Agent(name="task_a", model="gemini-2.0-flash", instruction="执行任务 A。")task_b =Agent(name="task_b", model="gemini-2.0-flash", instruction="执行任务 B。")
# 概念性：并发运行子智能体parallel_tasks =ParallelAgent(    name="concurrent_work",    sub_agents=[task_a, task_b])print("ADK：概念性 ParallelAgent 工作流程。")# results = parallel_tasks.run("输入数据") # 返回两者的结果

from autogen importAssistantAgent,UserProxyAgent,GroupChat,GroupChatManager
# 定义多个智能体...config_list = config_list_from_json("OAI_CONFIG_LIST")llm_config ={"config_list": config_list}coder =AssistantAgent(name="coder", llm_config=llm_config)pm =AssistantAgent(name="project_manager", llm_config=llm_config)tester =AssistantAgent(name="tester", llm_config=llm_config)user_proxy =UserProxyAgent(name="user", human_input_mode="NEVER")
groupchat =GroupChat(agents=[user_proxy, coder, pm, tester], messages=[], max_round=10)manager =GroupChatManager(groupchat=groupchat, llm_config=llm_config)
# user_proxy.initiate_chat(manager, message="开发并测试功能 X。")print("Autogen：概念性群聊设置。")

# 文件名：langchain_multi_2_conceptual.pyfrom typing importTypedDict,Annotatedfrom langgraph.graph importStateGraph,END
classAgentState(TypedDict):# 共享状态    input: str    result: str    sender: str
# 占位符智能体可运行对象（实现省略）def agent_1_runnable(state):return{"result":"A1","sender":"agent1"}def agent_2_runnable(state):return{"result":"A2","sender":"agent2"}
workflow =StateGraph(AgentState)workflow.add_node("agent1", agent_1_runnable)workflow.add_node("agent2", agent_2_runnable)workflow.set_entry_point("agent1")workflow.add_conditional_edges("agent1",lambda state: state['sender'],{"agent2":"agent2","__end__":END})workflow.add_edge('agent2','agent1')# 循环流app = workflow.compile()print("Langchain：概念性 LangGraph 结构，用于有状态的多智能体流。")

from crewai importAgent,Task,Crew,Processfrom langchain_openai importChatOpenAI
llm =ChatOpenAI(model="gpt-4o-mini")manager =Agent(role='管理者', goal='监督项目', backstory='...', llm=llm)# 管理者智能体worker1 =Agent(role='工作者1', goal='...', backstory='...', llm=llm)worker2 =Agent(role='工作者2', goal='...', backstory='...', llm=llm)
task_delegation =Task(description="为项目 X 分配子任务", agent=manager, expected_output="...")# 工作者任务可能动态创建或定义以供拾取
# 具有管理者智能体委托任务的团队hierarchical_crew =Crew(    agents=[manager, worker1, worker2],    tasks=[task_delegation],# 从管理者的任务开始    process=Process.hierarchical,# 启用委托流程    manager_llm=llm # 为管理者角色指定大语言模型)
# 顺序任务执行的团队# sequential_crew = Crew(#    agents=[researcher, writer],#    tasks=[task1, task2],#    process=Process.sequential # 任务按顺序运行# )
print("CrewAI：定义了具有层次流程的团队（管理者委托）。")# result = hierarchical_crew.kickoff()

from google.adk.agents importAgentimport datetime
def get_current_iso_time()-> str:"""返回当前日期和时间的 ISO 格式。"""# 文档字符串解释工具功能return datetime.datetime.now().isoformat()
time_agent =Agent(    name="time_agent", model="gemini-2.0-flash",    instruction="如果被问到，告知当前时间。",    tools=[get_current_iso_time]# 直接传递函数)print("ADK：工具定义为带有文档字符串的 Python 函数。")

# 概念性：ADK 可能提供类似于其他框架的装饰器from google.adk.tools import tool # 假设导入from google.adk.agents importAgent
@tool("weather_lookup")# 装饰器定义名称/模式def get_weather(city: str, unit: str ="celsius")-> str:"""获取城市天气。单位可以是 'celsius' 或 'fahrenheit'。"""# （API 调用实现）return f"{city} 的天气是晴天，25 {unit}。"
weather_agent =Agent(    name="weather_agent", model="gemini-2.0-flash",    instruction="提供天气预报。",    tools=[get_weather])
print("ADK：概念性工具定义，使用 @tool 装饰器。")

from autogen importAssistantAgent, config_list_from_jsonimport json
# 以 OpenAI 格式定义工具模式calculator_schema ={"type":"function","function":{"name":"calculator","description":"计算表达式的结果。","parameters":{"type":"object","properties":{"expression":{"type":"string","description":"数学表达式"}},"required":["expression"],},},}# 实际计算器函数需在其他地方可调用/注册def calculator(expression: str)-> str:return str(eval(expression))# 不安全的 eval
config_list = config_list_from_json("OAI_CONFIG_LIST")llm_config_tools ={"config_list": config_list,"tools":[calculator_schema]# 将模式传递到大语言模型配置}
calculator_agent =AssistantAgent(    name="calculator_agent",    llm_config=llm_config_tools)print("Autogen：通过 llm_config 中的 'tools' 列表配置工具，以实现大语言模型工具调用。")# 如果配置正确，智能体将直接输出工具调用

from langchain_community.tools importDuckDuckGoSearchRun# 导入特定工具# from langchain.agents import load_tools # 替代批量加载from langchain.agents importAgentExecutor# ... 其他导入from langchain_core.tools import tool
@tooldef get_word_length(word: str)->int:return len(word)
# 初始化预构建工具search =DuckDuckGoSearchRun()
# 组合自定义和预构建工具tools =[get_word_length, search]# agent_executor = AgentExecutor(..., tools=tools, ...)print("Langchain：使用预构建工具（DuckDuckGoSearchRun）结合自定义工具。")

from crewai_tools import tool # 从 crewai_tools 导入# 需要：pip install 'crewai[tools]'from crewai_tools importSerperDevTool# 示例预构建搜索工具from crewai importAgent# ... 其他导入
@tool("FileReadTool")def read_file(file_path: str)-> str:"""读取文件内容。"""try:with open(file_path,'r')as f:return f.read()exceptExceptionas e:return f"读取文件错误：{e}"
search_tool =SerperDevTool()
file_agent =Agent(# ... 其他参数 ...    tools=[read_file, search_tool])print("CrewAI：使用 crewai_tools 的 @tool 装饰器定义工具。")

# 文件名：adk_web_ui.pyfrom google.adk.agents importAgent
# 定义简单智能体agent =Agent(    name="chat_agent",    model="gemini-2.0-flash",    instruction="回答用户查询。")
# 保存此文件，然后在终端运行：# 1. 安装：pip install google-adk# 2. 运行：adk web --agent-file adk_web_ui.py# 访问 http://localhost:8080（默认端口）print("ADK：运行 'adk web --agent-file adk_web_ui.py' 以启动原生 Web UI。")

# 终端命令安装并运行 Autogen Studiopip install autogenstudioautogenstudio ui --port 8081# 访问 http://localhost:8081

# 文件名：crewai_streamlit_ui.py# 安装：pip install crewai streamlit langchain-openaifrom crewai importAgent,Task,Crewfrom langchain_openai importChatOpenAIimport streamlit as st
# 定义简单团队llm =ChatOpenAI(model="gpt-4o-mini")agent =Agent(role="助手", goal="协助用户", backstory="友好的 AI", llm=llm)task =Task(description="回答用户查询：{query}", agent=agent, expected_output="文本响应")crew =Crew(agents=[agent], tasks=[task])
# Streamlit UIst.title("CrewAI 助手")query = st.text_input("提问：")if query:    task.description = f"回答用户查询：{query}"    result = crew.kickoff()    st.write(result)
# 运行：streamlit run crewai_streamlit_ui.pyprint("CrewAI：运行 'streamlit run crewai_streamlit_ui.py' 以启动 UI。")