【Agent】框架开发实践(AutoGen、AgentScope、CAMEL、LangGraph)
note
- AutoGen 将复杂的协作抽象为一场由多角色参与的、可自动进行的“群聊”，其核心在于“以对话驱动协作”。
- AgentScope 则着眼于工业级应用的健壮性与可扩展性，为构建高并发、分布式的多智能体系统提供了坚实的工程基础。
- CAMEL 以其轻量级的“角色扮演”和“引导性提示”范式，展示了如何用最少的代码激发两个专家智能体之间深度、自主的协作。
- LangGraph 则回归到更底层的“状态机”模型，通过显式的图结构赋予开发者对工作流的精确控制，尤其是其循环能力，为构建可反思、可修正的智能体铺平了道路。
- langgraph用“图（Graph）”来描述 Agent 的执行流程，而不是用一坨 if-else
- LangGraph 是一个用于构建“可控、多步骤、状态化 LLM Agent”的流程编排框架，它用有向图来显式描述 Agent 的决策与执行路径。
文章目录
一、智能体框架
一个框架的本质，是提供一套经过验证的“规范”。它将所有智能体共有的、重复性的工作（如主循环、状态管理、工具调用、日志记录等）进行抽象和封装，让我们在构建新的智能体时，能够专注于其独特的业务逻辑，而非通用的底层实现。
智能体框架的必要性：
- 提升代码复用与开发效率：这是最直接的价值。一个好的框架会提供一个通用的
Agent基类或执行器，它封装了智能体运行的核心循环（Agent Loop）。无论是 ReAct 还是 Plan-and-Solve，都可以基于框架提供的标准组件快速搭建，从而避免重复劳动。 - 实现核心组件的解耦与可扩展性：一个健壮的智能体系统应该由多个松散耦合的模块组成。框架的设计会强制我们分离不同的关注点：
- 模型层 (Model Layer)：负责与大语言模型交互，可以轻松替换不同的模型（OpenAI, Anthropic, 本地模型）。
- 工具层 (Tool Layer)：提供标准化的工具定义、注册和执行接口，添加新工具不会影响其他代码。
- 记忆层 (Memory Layer)：处理短期和长期记忆，可以根据需求切换不同的记忆策略（如滑动窗口、摘要记忆）。 这种模块化的设计使得整个系统极具可扩展性，更换或升级任何一个组件都变得简单。
- 标准化复杂的状态管理：我们在
ReflectionAgent中实现的Memory类只是一个简单的开始。在真实的、长时运行的智能体应用中，状态管理是一个巨大的挑战，它需要处理上下文窗口限制、历史信息持久化、多轮对话状态跟踪等问题。一个框架可以提供一套强大而通用的状态管理机制，开发者无需每次都重新处理这些复杂问题。 - 简化可观测性与调试过程：当智能体的行为变得复杂时，理解其决策过程变得至关重要。一个精心设计的框架可以内置强大的可观测性能力。例如，通过引入事件回调机制（Callbacks），我们可以在智能体生命周期的关键节点（如
on_llm_start,on_tool_end,on_agent_finish）自动触发日志记录或数据上报，从而轻松地追踪和调试智能体的完整运行轨迹。这远比在代码中手动添加print语句要高效和系统化。
1、AutoGen框架
（1）优势和劣势
（1）优势
- 我们无需为智能体团队设计复杂的状态机或控制流逻辑，而是将一个完整的软件开发流程，自然地映射为产品经理、工程师和审查员之间的对话。这种方式更贴近人类团队的协作模式，显著降低了为复杂任务建模的门槛。开发者可以将更多精力聚焦于定义“谁（角色）”以及“做什么（职责）”，而非“如何做（流程控制）”。
- 框架允许通过系统消息（System Message）为每个智能体赋予高度专业化的角色。在案例中，ProductManager 专注于需求，而 CodeReviewer 则专注于质量。一个精心设计的智能体可以在不同项目中被复用，易于维护和扩展。
- 对于流程化任务，RoundRobinGroupChat 这样机制提供了清晰、可预测的协作流程。同时，UserProxyAgent 的设计为“人类在环”（Human-in-the-loop）提供了天然的接口。它既可以作为任务的发起者，也可以是流程的监督者和最终的验收者。这种设计确保了自动化系统始终处于人类的监督之下。
（2）局限性
- 虽然 RoundRobinGroupChat 提供了顺序化的流程，但基于 LLM 的对话本质上具有不确定性。智能体可能会产生偏离预期的回复，导致对话走向意外的分支，甚至陷入循环。
- 当智能体团队的工作结果未达预期时，调试过程可能非常棘手。与传统程序不同，我们得到的不是清晰的错误堆栈，而是一长串的对话历史。这被称为“对话式调试”的难题。
（2）相关例子
软件开发的流程是相对固定的（需求->编码->审查->测试），因此 RoundRobinGroupChat (轮询群聊) 是理想的选择。我们按照业务逻辑顺序，将四个智能体加入到参与者列表中。
2、AgentScope框架
在这个架构中，最底层是基础组件层 (Foundational Components)，它为整个框架提供了核心的构建块。Message 组件定义了统一的消息格式，支持从简单的文本交互到复杂的多模态内容；Memory 组件提供了短期和长期记忆管理；Model API 层抽象了对不同大语言模型的调用；而 Tool 组件则封装了智能体与外部世界交互的能力。
在基础组件之上，智能体基础设施层 (Agent-level Infrastructure) 提供了更高级的抽象。这一层不仅包含了各种预构建的智能体（如浏览器使用智能体、深度研究智能体），还实现了经典的 ReAct 范式，支持智能体钩子、并行工具调用、状态管理等高级特性。特别值得注意的是，这一层原生支持异步执行与实时控制，这是 AgentScope 相比其他框架的一个重要优势。
多智能体协作层 (Multi-Agent Cooperation) 是 AgentScope 的核心创新所在。MsgHub 作为消息中心，负责智能体间的消息路由和状态管理；而 Pipeline 系统则提供了灵活的工作流编排能力，支持顺序、并发等多种执行模式。这种设计使得开发者可以轻松构建复杂的多智能体协作场景。
最上层的开发与部署层 (Deployment & Devvelopment)则体现了 AgentScope 对工程化的重视。AgentScope Runtime 提供了生产级的运行时环境，而 AgentScope Studio 则为开发者提供了完整的可视化开发工具链。
优势和劣势：
通过这个"三国狼人杀"案例，我们深度体验了 AgentScope 框架的核心优势。该框架以其消息驱动的架构为核心，将复杂的游戏流程优雅地映射为一系列并发、异步的消息传递事件，从而避免了传统状态机的僵硬与复杂。结合其强大的结构化输出能力，我们将游戏规则直接转化为代码层面的约束，极大地提升了系统的稳定性和可预测性。这种设计范式不仅在性能上展现了其原生并发的优势，更在容错处理上保证了即使单个智能体出现异常，整体流程也能稳健运行。
然而，AgentScope 的工程化优势也带来了一定的复杂性成本。其消息驱动架构虽然强大，但对开发者的技术要求较高，需要理解异步编程、分布式通信等概念。对于简单的多智能体对话场景，这种架构可能显得过于复杂，存在"过度工程化"的风险。此外，作为相对较新的框架，其生态系统和社区资源还有待进一步完善。因此，AgentScope 更适合需要构建大规模、高可靠性的生产级多智能体系统，而对于快速原型开发或简单应用场景，选择更轻量级的框架可能更为合适。
3、CAMEL框架
1、优势
CAMEL 最大的优势在于其"轻架构、重提示"的设计哲学。相比 AutoGen 的复杂对话管理和 AgentScope 的分布式架构，CAMEL 通过精心设计的初始提示就能实现高质量的智能体协作。这种自然涌现的协作行为，往往比硬编码的工作流更加灵活和高效。
值得注意的是，CAMEL 框架正在经历快速的发展和演进。从其 GitHub 仓库 可以看到，CAMEL 已经远不止是一个简单的双智能体协作框架，目前已经具备：
多模态能力：支持文本、图像、音频等多种模态的智能体协作
工具集成：内置了丰富的工具库，包括搜索、计算、代码执行等
模型适配：支持 OpenAI、Anthropic、Google、开源模型等多种 LLM 后端
生态联动：与 LangChain、CrewAI、AutoGen 等主流框架实现了互操作性
2、主要局限性
（1）对提示工程的高度依赖
CAMEL 的成功很大程度上取决于初始提示的质量。这带来了几个挑战：
提示设计门槛：需要深入理解目标领域和 LLM 的行为特性
调试复杂性：当协作效果不佳时，很难定位是角色定义、任务描述还是交互规则的问题
一致性挑战：不同的 LLM 对相同提示的理解可能存在差异
（2）协作规模的限制
虽然 CAMEL 在双智能体协作上表现出色，但在处理大规模多智能体场景时面临挑战：
对话管理：缺乏像 AutoGen 那样的复杂对话路由机制
状态同步：没有 AgentScope 那样的分布式状态管理能力
冲突解决：当多个智能体意见分歧时，缺乏有效的仲裁机制
（3）任务适用性的边界
CAMEL 特别适合需要深度协作和创造性思维的任务，但在某些场景下可能不是最优选择：
严格流程控制：对于需要精确步骤控制的任务，LangGraph 的图结构更合适
大规模并发：AgentScope 的消息驱动架构在高并发场景下更有优势
复杂决策树：AutoGen 的群聊模式在多方决策场景下更加灵活
总的来说，CAMEL 代表了一种独特而优雅的多智能体协作范式。它通过"以人为本"的角色扮演设计，将复杂的系统工程问题转化为直观的人际协作模式。随着其生态系统的不断完善和功能的持续扩展，CAMEL 正在成为构建智能协作系统的重要选择之一。
4、LangGraph框架
1、优势
如我们的智能搜索助手案例所示，LangGraph 将一个完整的实时问答流程，显式地定义为一个由状态、节点和边构成的“流程图”。这种设计的最大优势是高度的可控性与可预测性。开发者可以精确地规划智能体的每一步行为，这对于构建需要高可靠性和可审计性的生产级应用至关重要。其最强大的特性在于对循环（Cycles）的原生支持。通过条件边，我们可以轻松构建“反思-修正”循环，例如在我们的案例中，如果搜索失败，可以设计一个回退到备用方案的路径。这是构建能够自我优化和具备容错能力的智能体的关键。
此外，由于每个节点都是一个独立的 Python 函数，这带来了高度的模块化。同时，在流程中插入一个等待人类审核的节点也变得非常直接，为实现可靠的“人机协作”（Human-in-the-loop）提供了坚实的基础。
2、局限性
与基于对话的框架相比，LangGraph 需要开发者编写更多的前期代码（Boilerplate）。定义状态、节点、边等一系列操作，使得对于简单任务而言，开发过程显得更为繁琐。开发者需要更多地思考“如何控制流程（how）”，而不仅仅是“做什么（what）”。由于工作流是预先定义的，LangGraph 的行为虽然可控，但也缺少了对话式智能体那种动态的、“涌现”式的交互。它的强项在于执行一个确定的、可靠的流程，而非模拟开放式的、不可预测的社会性协作。
调试过程同样存在挑战。虽然流程比对话历史更清晰，但问题可能出在多个环节：某个节点内部的逻辑错误、在节点间传递的状态数据发生异变，或是边跳转的条件判断失误。这要求开发者对整个图的运行机制有全局性的理解。
实践部分：基于 LangGraph + Tavily API 的真实搜索系统
代码流程：understand → search → answer
1、理解问题：understand_query_node，将query转为相关搜索关键词
2、搜索：tavily_search_node，用1的搜索词进行网络搜索，返回标题+摘要+链接
3、生成答案：generate_answer_node，根据搜索内容用llm生成答案
"""
智能搜索助手 - 基于 LangGraph + Tavily API 的真实搜索系统
1. 理解用户需求
2. 使用Tavily API真实搜索信息
3. 生成基于搜索结果的回答
"""
import asyncio
from typing import TypedDict, Annotated
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
from langchain_community.chat_models import ChatZhipuAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
import os
from dotenv import load_dotenv
from tavily import TavilyClient
# 加载环境变量
load_dotenv()
# 定义状态结构
class SearchState(TypedDict):
messages: Annotated[list, add_messages]
user_query: str # 用户查询
search_query: str # 优化后的搜索查询
search_results: str # Tavily搜索结果
final_answer: str # 最终答案
step: str # 当前步骤
# 初始化模型和Tavily客户端
# llm = ChatOpenAI(
# model=os.getenv("LLM_MODEL_ID", "gpt-4o-mini"),
# api_key=os.getenv("LLM_API_KEY"),
# base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"),
# temperature=0.7
# )
llm = ChatZhipuAI(
model=os.getenv("LLM_MODEL_ID"),
api_key=os.getenv("LLM_API_KEY"),
base_url=os.getenv("LLM_BASE_URL"),
temperature=0.7
)
# 初始化Tavily客户端
tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
def understand_query_node(state: SearchState) -> SearchState:
"""步骤1：理解用户查询并生成搜索关键词"""
# 获取最新的用户消息
user_message = ""
for msg in reversed(state["messages"]):
if isinstance(msg, HumanMessage):
user_message = msg.content
break
understand_prompt = f"""分析用户的查询："{user_message}"
请完成两个任务：
1. 简洁总结用户想要了解什么
2. 生成最适合搜索的关键词（中英文均可，要精准）
格式：
理解：[用户需求总结]
搜索词：[最佳搜索关键词]"""
print(f"=======给大模型的prompt:\n{understand_prompt}")
# response = llm.invoke([SystemMessage(content=understand_prompt)])
response = llm.invoke([HumanMessage(content=understand_prompt)])
# 提取搜索关键词
response_text = response.content
search_query = user_message # 默认使用原始查询
if "搜索词：" in response_text:
search_query = response_text.split("搜索词：")[1].strip()
elif "搜索关键词：" in response_text:
search_query = response_text.split("搜索关键词：")[1].strip()
return {
"user_query": response.content,
"search_query": search_query,
"step": "understood",
"messages": [AIMessage(content=f"我理解您的需求：{response.content}")]
}
def tavily_search_node(state: SearchState) -> SearchState:
"""步骤2：使用Tavily API进行真实搜索"""
search_query = state["search_query"]
try:
print(f"🔍 正在搜索: {search_query}")
# 调用Tavily搜索API
response = tavily_client.search(
query=search_query,
search_depth="basic",
include_answer=True,
include_raw_content=False,
max_results=5
)
# 处理搜索结果
search_results = ""
# 优先使用Tavily的综合答案
if response.get("answer"):
search_results = f"综合答案：\n{response['answer']}\n\n"
# 添加具体的搜索结果
if response.get("results"):
search_results += "相关信息：\n"
for i, result in enumerate(response["results"][:3], 1):
title = result.get("title", "")
content = result.get("content", "")
url = result.get("url", "")
search_results += f"{i}. {title}\n{content}\n来源：{url}\n\n"
if not search_results:
search_results = "抱歉，没有找到相关信息。"
return {
"search_results": search_results,
"step": "searched",
"messages": [AIMessage(content=f"✅ 搜索完成！找到了相关信息，正在为您整理答案...")]
}
except Exception as e:
error_msg = f"搜索时发生错误: {str(e)}"
print(f"❌ {error_msg}")
return {
"search_results": f"搜索失败：{error_msg}",
"step": "search_failed",
"messages": [AIMessage(content="❌ 搜索遇到问题，我将基于已有知识为您回答")]
}
def generate_answer_node(state: SearchState) -> SearchState:
"""步骤3：基于搜索结果生成最终答案"""
# 检查是否有搜索结果
if state["step"] == "search_failed":
# 如果搜索失败，基于LLM知识回答
fallback_prompt = f"""搜索API暂时不可用，请基于您的知识回答用户的问题：
用户问题：{state['user_query']}
请提供一个有用的回答，并说明这是基于已有知识的回答。"""
# response = llm.invoke([SystemMessage(content=fallback_prompt)])
# HumanMessage
response = llm.invoke([HumanMessage(content=fallback_prompt)])
return {
"final_answer": response.content,
"step": "completed",
"messages": [AIMessage(content=response.content)]
}
# 基于搜索结果生成答案
answer_prompt = f"""基于以下搜索结果为用户提供完整、准确的答案：
用户问题：{state['user_query']}
搜索结果：
{state['search_results']}
请要求：
1. 综合搜索结果，提供准确、有用的回答
2. 如果是技术问题，提供具体的解决方案或代码
3. 引用重要信息的来源
4. 回答要结构清晰、易于理解
5. 如果搜索结果不够完整，请说明并提供补充建议"""
# response = llm.invoke([SystemMessage(content=answer_prompt)])
# HumanMessage
response = llm.invoke([HumanMessage(content=answer_prompt)])
return {
"final_answer": response.content,
"step": "completed",
"messages": [AIMessage(content=response.content)]
}
# 构建搜索工作流
def create_search_assistant():
workflow = StateGraph(SearchState)
# 添加三个节点
workflow.add_node("understand", understand_query_node)
workflow.add_node("search", tavily_search_node)
workflow.add_node("answer", generate_answer_node)
# 设置线性流程
workflow.add_edge(START, "understand")
workflow.add_edge("understand", "search")
workflow.add_edge("search", "answer")
workflow.add_edge("answer", END)
# 编译图
memory = InMemorySaver()
app = workflow.compile(checkpointer=memory)
return app
async def main():
"""主函数：运行智能搜索助手"""
# 检查API密钥
if not os.getenv("TAVILY_API_KEY"):
print("❌ 错误：请在.env文件中配置TAVILY_API_KEY")
return
app = create_search_assistant()
print("🔍 智能搜索助手启动！")
print("我会使用Tavily API为您搜索最新、最准确的信息")
print("支持各种问题：新闻、技术、知识问答等")
print("(输入 'quit' 退出)\n")
session_count = 0
while True:
user_input = input("🤔 您想了解什么: ").strip()
if user_input.lower() in ['quit', 'q', '退出', 'exit']:
print("感谢使用！再见！👋")
break
if not user_input:
continue
session_count += 1
config = {"configurable": {"thread_id": f"search-session-{session_count}"}}
# 初始状态
initial_state = {
"messages": [HumanMessage(content=user_input)],
"user_query": "",
"search_query": "",
"search_results": "",
"final_answer": "",
"step": "start"
}
try:
print("\n" + "="*60)
print(f"check 代码运行到try里面的地方\n")
# 执行工作流
async for output in app.astream(initial_state, config=config):
for node_name, node_output in output.items():
print(f"check 代码运行到工作流的第一个for循环中\n")
if "messages" in node_output and node_output["messages"]:
latest_message = node_output["messages"][-1]
if isinstance(latest_message, AIMessage):
if node_name == "understand":
print(f"🧠 理解阶段: {latest_message.content}")
elif node_name == "search":
print(f"🔍 搜索阶段: {latest_message.content}")
elif node_name == "answer":
print(f"\n💡 最终回答:\n{latest_message.content}")
print("\n" + "="*60 + "\n")
except Exception as e:
print(f"❌ 发生错误: {e}")
print("请重新输入您的问题。\n")
if __name__ == "__main__":
asyncio.run(main())
langgraph框架的节点函数相关写法：
def 节点函数(state: YourState) -> dict:
# 1. 入参：接收完整的state
# 2. 返回：只返回需要更新的字段（不用返回全部）
return {
"某个字段": "新值" # 会自动合并到state中
}
Reference
[1] https://datawhalechina.github.io/hello-agents/#/./chapter6/%E7%AC%AC%E5%85%AD%E7%AB%A0%20%E6%A1%86%E6%9E%B6%E5%BC%80%E5%8F%91%E5%AE%9E%E8%B7%B5
[2] https://docs.langchain.com/oss/python/integrations/chat/zhipuai
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
20
0- 0
已为社区贡献9条内容
所有评论(0)