这篇文章旨在让这个选择变得更容易一些。在过去的几周里,我使用当前主流的人工智能开发框架构建了相同的LLM代理,以便在技术层面检查每个框架的优缺点。

 

本文旨在帮助你在使用自己定制的LLM代理还是使用现有LLM代理框架之间作出正确的选择。

简介

首先,要感谢John Gilhuly对本文的贡献。

当下,人工智能代理暂时处在大休整时期。随着多个新的AI开发框架的不断出现和人们对该领域不断进行新的投资,现代人工智能代理正在克服不稳定的初始阶段,迅速取代RAG而成为实施重点。那么,2024年最终会成为什么样的年份呢?是自主人工智能系统接管我们人工来书写电子邮件、预订航班、处理数据,还是与任何其他年份一样以相似方式执行上述任务呢?

也许情况与前者一样,但是要达到这种程度还有很多工作要做。任何构建LLM代理的开发人员不仅必须选择基础开发设施——使用哪种模型、使用场景和架构——还必须选择要利用哪种开发框架。你是选择长期使用的LangGraph,还是新进入市场的LlamaIndex工作流?还是你走传统路线,自己编写整个代码呢?

这篇文章旨在让这个选择变得更容易一些。在过去的几周里,我使用当前主流的人工智能开发框架构建了相同的LLM代理,以便在技术层面检查每个框架的优缺点。本文中涉及的每个代理的所有源代码都可以在仓库地址处找到。

LLM代理类型

当前,业界主要用于测试目的的LLM代理开发涉及到很多方面的内容,例如函数调用、多种相关工具或技能、与外部资源的连接以及共享状态或内存,等等。

归纳起来看,几乎所有LLM代理都具有以下功能:

  • 回答知识库中的问题。
  • 与数据对话:回答有关LLM应用程序遥测数据的问题。
  • 分析数据:分析检索到的遥测数据中的高级趋势和模式。

为了完成这些任务,LLM代理需要具备三个基础技能:使用产品文档的RAG、在跟踪数据库上生成SQL和数据分析。一种典型的实现方案是,使用开源的Python包Gradio来快速构建一个代理用户界面,而LLM代理本身被构造为聊天机器人。

基于定制代码的代理(无框架方案)

开发LLM代理时,你的第一个选择很可能是完全跳过市场上现有框架,而完全由自己来构建一个代理。在最开始着手做这种项目时,这是我采用的方法。

纯代码架构

下面展示的基于代码的代理是由一个OpenAI驱动的路由器组成的,该路由器使用函数调用来选择要使用的正确技能。该技能完成后,它将返回路由器以调用另一个技能或者是对用户作出响应。

在这个代理中,始终保持一个持续的消息和响应列表,在每次调用时将其完全传递到路由器中,以便在循环中保留相应的上下文信息。

复制

def router(messages): if not any( isinstance(message, dict) and message.get(“role”) == “system” for message in messages ): system_prompt = {“role”: “system”, “content”: SYSTEM_PROMPT} messages.append(system_prompt) response = client.chat.completions.create( model=”gpt-4o”, messages=messages, tools=skill_map.get_combined_function_description_for_openai(), ) messages.append(response.choices[0].message) tool_calls = response.choices[0].message.tool_calls if tool_calls: handle_tool_calls(tool_calls, messages) return router(messages) else: return response.choices[0].message.content

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.

技能本身是在自己的类中定义的(例如GenerateSQLQuery),而所有这些技能信息共同保存在SkillMap类中。路由器本身只与SkillMap类交互,它使用SkillMap类实现来加载技能名称、描述和可调用函数。这种方法意味着,向代理添加新技能就像将该技能编写为自己的类一样简单,然后将其添加到SkillMap类中的技能列表中。这里的想法是,在不干扰路由器代码的情况下轻松添加新技能。

复制

class SkillMap: def __init__(self): skills = [AnalyzeData(), GenerateSQLQuery()] self.skill_map = {} for skill in skills: self.skill_map[skill.get_function_name()] = ( skill.get_function_dict(), skill.get_function_callable(), ) def get_function_callable_by_name(self, skill_name) -> Callable: return self.skill_map[skill_name][1] def get_combined_function_description_for_openai(self): combined_dict = [] for _, (function_dict, _) in self.skill_map.items(): combined_dict.append(function_dict) return combined_dict def get_function_list(self): return list(self.skill_map.keys()) def get_list_of_function_callables(self): return [skill[1] for skill in self.skill_map.values()] def get_function_description_by_name(self, skill_name): return str(self.skill_map[skill_name][0][“function”])

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.

总体而言,这种方法实施起来相当简单,不过也存在一些挑战。

纯代码代理方案的挑战

第一个难点在于构建路由器系统提示。通常,上述示例中的路由器坚持自己生成SQL,而不是将其委托给合适的技能。如果你曾经试图不让LLM做某事,你就会知道这种经历有多么令人沮丧;找到一个可用的提示需要进行多轮调试。考虑到每个步骤的不同输出格式也是很棘手的。由于我选择不使用结构化输出,因此我必须为路由器和技能中每个LLM调用的多种不同格式做好准备。

纯代码代理方案的优点

基于代码的方法提供了一个很好的基础架构和起点,提供了一种学习代理如何工作的好方法,而不是依赖于主流框架中的现成的代理教程。尽管说服LLM的行为可能具有挑战性,但代码结构本身足够简单,可以使用,并且可能对某些使用场景也极有意义。有关这些使用场景的更多信息,请参阅接下来的分析。

LangGraph

LangGraph是历史最悠久的代理框架之一,于2024年1月首次发布。该框架旨在通过采用Pregel图结构来解决现有管道和链的非循环性。LangGraph通过添加节点、边和条件边的概念来遍历图,使得在代理中定义循环变得更加容易。LangGraph构建在LangChain之上,并使用LangChain框架中的对象和类型。

LangGraph架构

LangGraph代理看起来与其原论文中的基于代码的代理相似,但它后台的实现代码却截然不同。LangGraph在技术上仍然使用“路由器”,因为它通过函数调用OpenAI,并使用响应继续进行新的步骤。然而,程序在技能之间移动的方式完全不同。

复制

tools = [generate_and_run_sql_query, data_analyzer] model = ChatOpenAI(model=”gpt-4o”, temperature=0).bind_tools(tools) def create_agent_graph(): workflow = StateGraph(MessagesState) tool_node = ToolNode(tools) workflow.add_node(“agent”, call_model) workflow.add_node(“tools”, tool_node) workflow.add_edge(START, “agent”) workflow.add_conditional_edges( “agent”, should_continue, ) workflow.add_edge(“tools”, “agent”) checkpointer = MemorySaver() app = workflow.compile(checkpointer=checkpointer) return app

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.

这里定义的图有一个用于初始OpenAI调用的节点,上面称为“agent”,还有一个用于工具处理步骤的节点,称为“tools”。LangGraph提供了一个名为ToolNode的内置对象,它负责获取可调用工具的列表,并根据ChatMessage响应触发它们,然后再次返回“agent”节点。

复制

def should_continue(state: MessagesState): messages = state[“messages”] last_message = messages[-1] if last_message.tool_calls: return “tools” return END def call_model(state: MessagesState): messages = state[“messages”] response = model.invoke(messages) return {“messages”: [response]}

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.

在每次调用“agent”节点(换句话说:基于代码的代理中的路由器)后,should_concontinue边决定是将响应返回给用户还是传递给ToolNode来处理工具调用。

在每个节点中,“state”存储来自OpenAI的消息和响应列表,这一点类似于基于代码的代理的方法。

LangGraph方案的挑战

示例中LangGraph实现的大部分困难在于LangChain对象的使用,此方案需要借助这个对象来使事情顺利进行。

挑战#1:函数调用验证

为了使用ToolNode对象,我必须重构我现有的大部分Skill代码。ToolNode接受一个可调用函数列表,这最初让我认为我可以使用现有的函数,但由于我设计的函数参数方面的原因,事情发生了一些变化。

这些技能都被定义为具有可调用成员函数的类。这意味着,它们的第一个参数是“self”。GPT-4o足够聪明,不会在生成的函数调用中包含“self”参数,但不幸的是LangGraph将其解读为由于缺少参数而导致的验证错误。

这花了我几个小时才弄清楚,因为错误消息将函数中的第三个参数(数据分析技能上的“args”)标记为缺少的参数:

复制

pydantic.v1.error_wrappers.ValidationError: 1 validation error for data_analysis_toolSchema args field required (type=value_error.missing)

  • 1.
  • 2.

值得一提的是,错误消息来自Pydantic,而不是LangGraph。

最终,我咬紧牙关,用Langchain的@tool装饰器将我的技能重新定义为一些基本方法,终于使得代码正常进行。

复制

@tool def generate_and_run_sql_query(query: str): “””根据提示符生成并运行一个SQL查询。 参数: query (str): 一个包含原始用户提示符的字符串。 返回值: str: SQL查询的结果。 “””

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.

挑战#2:调试

正如前文所述,在框架内进行调试是颇为困难的事情。这主要归结为令人困惑的错误消息和抽象概念,使查看变量变得更加困难。

抽象概念主要出现在尝试调试代理周围发送的消息时。LangGraph将这些消息存储在状态[“messages”]中。图中的一些节点会自动从这些消息中提取信息,这可能会使节点访问消息时难以理解消息的含义。

代理动作的顺序视图

LangGraph方案的优点

LangGraph的主要优点之一是易于使用,因为图形结构代码干净且易于访问。特别是如果你有复杂的节点逻辑时,只通过图的单一视图有助于更容易地理解代理是如何连接在一起的。LangGraph方案也使得转换现有的基于LangChain构建的应用程序变得非常简单。

小结

如果你仅使用LangGraph框架中的所有内容,那么LangGraph会顺利地工作。但是,如果你还想使用框架外的一些内容进行开发的话,那么你需要为调试一些难题做好准备。

LlamaIdex工作流

工作流是LLM代理框架领域的新方案,于今年夏天早些时候首次亮相。与LangGraph一样,它旨在使循环代理更容易构建。另外,LLM工作流还特别关注异步运行方式。

LLM工作流的一些元素似乎是对LangGraph的直接响应,特别是它使用事件而不是边和条件边。工作流使用步骤(类似于LangGraph中的节点)来容纳逻辑,并且在步骤之间发送和接收事件。

上面的结构看起来与LangGraph结构相似,只是增加了一点内容。我在工作流中添加了一个设置步骤来负责准备代理上下文,下面将进一步介绍这方面内容。值得注意的是,尽管这两种方案的结构相似,但是驱动它们的代码却大不相同。

工作流架构

下面的代码定义了一个工作流结构。与LangGraph方案中代码类似,这是我准备状态并将技能附加到LLM对象的地方。

复制

class AgentFlow(Workflow): def __init__(self, llm, timeout=300): super().__init__(timeout=timeout) self.llm = llm self.memory = ChatMemoryBuffer(token_limit=1000).from_defaults(llm=llm) self.tools = [] for func in skill_map.get_function_list(): self.tools.append( FunctionTool( skill_map.get_function_callable_by_name(func), metadata=ToolMetadata( name=func, description=skill_map.get_function_description_by_name(func) ), ) ) @step async def prepare_agent(self, ev: StartEvent) -> RouterInputEvent: user_input = ev.input user_msg = ChatMessage(role=”user”, content=user_input) self.memory.put(user_msg) chat_history = self.memory.get() return RouterInputEvent(input=chat_history)

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.

这也是我定义额外步骤“prepare_agent”的地方。此步骤根据用户输入创建ChatMessage并将其添加到工作流内存中。将其拆分为一个单独的步骤意味着,当代理循环执行步骤时,我们确实会返回它,这样就避免了将用户消息重复添加到内存中。

在LangGraph的例子中,我用一个位于图外的run_agent方法完成了同样的事情。这种变化主要是风格上的,但在我看来,像我们在这里所做的那样,用工作流和图形来容纳这种逻辑会更清晰。

设置好工作流后,我定义了路由代码:

复制

@step async def router(self, ev: RouterInputEvent) -> ToolCallEvent | StopEvent: messages = ev.input if not any( isinstance(message, dict) and message.get(“role”) == “system” for message in messages ): system_prompt = ChatMessage(role=”system”, content=SYSTEM_PROMPT) messages.insert(0, system_prompt) with using_prompt_template(template=SYSTEM_PROMPT, version=”v0.1″): response = await self.llm.achat_with_tools( model=”gpt-4o”, messages=messages, tools=self.tools, ) self.memory.put(response.message) tool_calls = self.llm.get_tool_calls_from_response(response, error_on_no_tool_call=False) if tool_calls: return ToolCallEvent(tool_calls=tool_calls) else: return StopEvent(result=response.message.content)

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.

以及工具调用处理代码:

复制

@step async def tool_call_handler(self, ev: ToolCallEvent) -> RouterInputEvent: tool_calls = ev.tool_calls for tool_call in tool_calls: function_name = tool_call.tool_name arguments = tool_call.tool_kwargs if “input” in arguments: arguments[“prompt”] = arguments.pop(“input”) try: function_callable = skill_map.get_function_callable_by_name(function_name) except KeyError: function_result = “Error: Unknown function call” function_result = function_callable(arguments) message = ChatMessage( role=”tool”, content=function_result, additional_kwargs={“tool_call_id”: tool_call.tool_id}, ) self.memory.put(message) return RouterInputEvent(input=self.memory.get())

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.

上面两部分代码看起来都比LangGraph代理更类似于基于代码的代理。这主要是因为工作流将条件路由逻辑保留在步骤中,而不是保留在条件边中——其中的部分代码行以前是对应于LangGraph中的条件边,而现在它们只是路由步骤的一部分——而且LangGraph有一个ToolNode对象,它几乎可以自动执行tool_call_handler方法中的所有操作。

经过路由步骤,我很高兴看到的一件事是,我可以将我的SkillMap和基于代码的代理中的现有技能与工作流一起使用。这些不需要更改即可使用工作流,这让我的工作变得更加轻松。

工作流程方案的挑战

挑战#1:同步与异步

虽然异步执行更适合实时代理,但调试同步代理要容易得多。工作流被设计为异步工作;因此,试图强制同步执行变得非常困难。

我最初以为我可以删除“async”方法名称,并从“achat_with_tools”切换到“chat_with_tools”。然而,由于Workflow类中的底层方法也被标记为异步,因此有必要重新定义这些方法以便同步运行。我最终坚持使用异步方法,但这并没有使调试变得更加困难。

代理动作的顺序视图

在LangGraph方案的困境重演过程中,围绕技能上令人困惑的Pydantic验证错误出现了类似的问题。幸运的是,这次这些问题更容易解决,因为工作流能够很好地处理成员函数。最终,我不得不更加规范地为我的技能创建LlamaIndex FunctionTool对象:

复制

for func in skill_map.get_function_list(): self.tools.append(FunctionTool( skill_map.get_function_callable_by_name(func), metadata=ToolMetadata(name=func, description=skill_map.get_function_description_by_name(func))))

  • 1.
  • 2.
  • 3.
  • 4.

这段代码摘自构建FunctionTools工具类的函数AgentFlow__init__。

工作流方案的优点

我构建工作流代理比构建LangGraph代理容易得多,主要是因为工作流仍然要求我自己编写路由逻辑和工具处理代码,而不是提供内置函数。这也意味着,我的工作流代理看起来与我的基于代码的代理非常相似。

最大的区别在于事件的使用。我使用了两个自定义事件在代理中的步骤之间移动:

复制

class ToolCallEvent(Event): tool_calls: list[ToolSelection] class RouterInputEvent(Event): input: list[ChatMessage]

  • 1.
  • 2.
  • 3.
  • 4.
  • 5.

基于事件的“发射器-接收器”架构取代了直接调用代理中的一些方法,如工具调用处理程序。

如果你有更复杂的系统,其中有多个异步触发的步骤的话,可能会发出多个事件,那么这种架构对于清晰地管理这些步骤非常有帮助。

工作流的其他好处包括:此方案非常轻量级,不会给你强加太多的结构(除了使用某些LlamaIdex对象),而且它基于事件的架构为直接函数调用提供了一种有用的替代方案,特别是对于复杂的异步应用程序而言。

框架比较

纵观上述三种方法,每种方法都各有其优点。

无框架方法是最容易实现的。因为任何抽象都是由开发人员自己定义的(即上例中的SkillMap对象),所以保持各种类型和对象的简洁是很容易的。然而,代码的可读性和可访问性完全取决于单个开发人员。很容易看出,在没有采用一些强制的结构定义的情况下,越来越复杂的代理会变得一团糟。

LangGraph框架本身提供了相当多的结构,这使得代理的定义非常清晰。如果一个更大的团队正在合作开发一个代理,这种结构将提供一种有助于实施架构的方法。LangGraph也可能为那些不熟悉结构的人提供一个很好的代理起点。然而,有一个权衡——由于LangGraph为你做了很多工作,如果你不完全接受该框架,可能会感觉有些头痛;代码可能非常干净,但你可能会为其付出更多的调试代价。

在上述三种方法中,工作流方案位于中间。基于事件的架构可能对某些项目非常有帮助;事实上,在使用LlamaIdex类型方面所需的编码量更少,这为那些在应用程序中没有完全使用过该框架的人提供了更大的灵活性。

最终,核心问题可能归结为“你已经在使用LlamaIndex或LangChain来编排你的应用程序了吗?”LangGraph和工作流都与各自的底层框架紧密相连,以至于每个特定于代理的框架的额外好处可能不会让你只凭优点来切换它们。

不过,纯代码方法可能永远是一个具有吸引力的选择。如果你有足够的严谨性来记录和执行任何创建的抽象的话,那么就很容易确保外部框架中没有任何设置会减缓你的开发速度。

选择代理框架的关键问题

当然,“视情况而定”从来不是一个令人满意的答案。下面的三个问题可以帮助你决定在下一个代理项目中使用哪个框架。

  • 你是否已经在项目的重要部分使用LlamaIndex或LangChain?

如果是,请先优先分析这种选择方案。

  • 你熟悉常见的代理结构吗?还是想知道你应该如何构建代理?

如果你大致属于后一种情形,请尝试工作流方案。不过,如果你真的属于后一种情形,试试LangGraph方案吧。

  • 以前构建过你自己的代理吗?

框架方案的好处之一是,每个框架都有许多教程和示例。相比而言,可用于构建纯代码代理的示例代码要少得多。

结论

无论如何,选择一个代理框架只是影响生成式人工智能系统生产结果的众多选择之一。与往常一样,构建强大的保护措施并进行LLM跟踪是非常值得推荐的做法,并且随着新的代理框架、研究成果和模型不断颠覆既定技术,这样做也变得更为机动灵活。

译者介绍

朱先忠,51CTO社区编辑,51CTO专家博客、讲师,潍坊一所高校计算机教师,自由编程界老兵一枚。

Loading

作者 yinhua

发表回复