用于测试的 Agent 包括函数调用、多种工具或技能、与外部资源的连接以及共享状态或内存。

Agent 具有以下功能：

• 回答知识库中的问题

• 与数据对话：回答有关 LLM 应用程序遥测数据的问题

• 分析数据：分析检索到的遥测数据中的更高级别趋势和模式

为了实现这些，Agent 有三种起始技能：带有产品文档的 RAG、跟踪数据库上的 SQL 生成和数据分析。Agent  UI 使用简单的 gradio 驱动界面，Agent 本身构造为聊天机器人。

开发 Agent 时，第一个选择是完全跳过框架，完全自己构建 Agent。在着手这个项目时，我采用的就是这种方法。

纯代码架构

下面的基于代码的 Agent 由 OpenAI 驱动的路由器组成，该路由器使用函数调用来选择要使用的正确技能。该技能完成后，它会返回路由器以调用另一项技能或响应用户。

Agent 会保存一个持续的消息和响应列表，这些列表在每次调用时都会完全传递到路由器中，以在整个周期中保留上下文。

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_callsif tool_calls:handle_tool_calls(tool_calls, messages)return router(messages)else:return response.choices[0].message.content

技能本身在自己的类（例如 GenerateSQLQuery）中定义，这些类共同保存在 SkillMap 中。路由器本身只与 SkillMap 交互，它使用 SkillMap 加载技能名称、描述和可调用函数。这种方法意味着向 Agent 添加新技能就像将该技能写为其自己的类，然后将其添加到 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"])

总体而言，这种方法实施起来相当简单，但也面临一些挑战。

纯代码 Agent 的挑战

第一个困难在于构造路由器系统提示。通常，上面示例中的路由器坚持自己生成 SQL，而不是将其委托给正确的技能。如果你曾经尝试让 LLM 不做某事，你就会知道这种体验有多么令人沮丧；找到一个有效的提示需要经过多轮调试。考虑每个步骤的不同输出格式也很棘手。由于我选择不使用结构化输出，因此我必须准备好应对路由器和技能中每个 LLM 调用的多种不同格式。

纯代码 Agent 的好处

基于代码的方法提供了良好的基线和起点，提供了一种很好的方法来学习 Agent 的工作原理，而无需依赖来自主流框架的固定 Agent 教程。虽然说服 LLM 表现可能具有挑战性，但代码结构本身足够简单易用，可能对某些用例有意义（更多信息请参见下面的分析部分）。

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

LangGraph 架构

LangGraph Agent 在纸面上看起来与基于代码的 Agent 类似，但其背后的代码却截然不同。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

这里定义的图表有一个用于初始 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]}

每次调用“Agent”节点（换句话说：基于代码的 Agent 中的路由器）后，should_continue 边缘都会决定是否将响应返回给用户或传递给 ToolNode 来处理工具调用。

在每个节点中，“状态”存储来自 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_toolSchemaargs field required (type=value_error.missing)

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

我最终下定决心，用 Langchain 的 @tool 装饰器将我的技能重新定义为基本方法，并让一切正常运转。

@tooldef generate_and_run_sql_query(query: str):"""Generates and runs an SQL query based on the prompt.
Args:query (str): A string containing the original user prompt.
Returns:str: The result of the SQL query."""

挑战 #2：调试

如前所述，在框架中调试很困难。这主要归结于令人困惑的错误消息和抽象概念，这些使查看变量变得更加困难。

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

LangGraph 的优势

LangGraph 的主要优势之一是易于使用。图形结构代码简洁易懂。特别是如果你有复杂的节点逻辑，拥有图形的单一视图可以更轻松地理解 Agent 是如何连接在一起的。LangGraph 还可以轻松转换在 LangChain 中构建的现有应用程序。

总结

如果你使用框架中的所有内容，LangGraph 可以干净利落地运行；如果你不使用它，请准备好面对一些调试难题。

Workflows 是 Agent 框架领域的新成员，于今年夏初首次亮相。与 LangGraph 一样，它旨在使循环代理更易于构建。Workflows 还特别注重异步运行。

Workflows 的一些元素似乎直接响应了 LangGraph，特别是它使用事件而不是边和条件边。Workflows 使用步骤（类似于 LangGraph 中的节点）来容纳逻辑，并使用发出和接收事件在步骤之间移动。

上面的结构看起来与 LangGraph 结构类似，除了一个附加项。我在 Workflow 中添加了一个设置步骤来准备 Agent 上下文，下面将详细介绍。尽管结构相似，但支持它的代码却大不相同。

Workflows 架构

下面的代码定义了 Workflow 结构。与 LangGraph 类似，这是我准备状态并将技能附加到 LLM 对象的地方。

class AgentFlow(Workflow):def __init__(self, llm, timeout=300):super().__init__(timeout=timeout)self.llm = llmself.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)),))
@stepasync def prepare_agent(self, ev: StartEvent) -> RouterInputEvent:user_input = ev.inputuser_msg = ChatMessage(role="user", content=user_input)self.memory.put(user_msg)
chat_history = self.memory.get()return RouterInputEvent(input=chat_history)

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

在 LangGraph 案例中，我使用位于图表之外的 run_agent 方法完成了同样的事情。这种变化主要是风格上的，但我认为像我们在这里所做的那样，将此逻辑与工作流和图表放在一起会更简洁。

设置好工作流后，我定义了路由代码：

@stepasync 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)

以及工具调用处理代码：

@stepasync def tool_call_handler(self, ev: ToolCallEvent) -> RouterInputEvent:tool_calls = ev.tool_calls
for tool_call in tool_calls:function_name = tool_call.tool_namearguments = tool_call.tool_kwargsif "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())

这两个看起来更像基于代码的 Agent 而不是 LangGraph Agent 。这主要是因为 Workflows 将条件路由逻辑保留在步骤中而不是条件边缘中 — 第 18-24 行是 LangGraph 中的条件边缘，而现在它们只是路由步骤的一部分 — 并且 LangGraph 有一个 ToolNode 对象，它几乎自动执行 tool_call_handler 方法中的所有操作。

经过路由步骤后，我很高兴看到一件事，那就是我可以将我的 SkillMap 和基于代码的 Agent 中的现有技能与 Workflows 结合使用。这些不需要任何更改即可与 Workflows 配合使用，这让我的生活变得轻松多了。

Workflows 的挑战

挑战 #1：同步与异步

虽然异步执行对于实时 Agent 来说是更好的选择，但调试同步 Agent 要容易得多。Workflows 旨在异步工作，尝试强制同步执行非常困难。

我最初以为我只需删除“异步”方法标识，并从“achat_with_tools”切换到“chat_with_tools”即可。但是，由于 Workflow 类中的底层方法也被标记为异步，因此必须重新定义这些方法才能同步运行。我最终坚持使用异步方法，但这并没有使调试变得更加困难。

挑战 #2：Pydantic 验证错误

与 LangGraph 的困境如出一辙，类似的问题也出现在技能上令人困惑的 Pydantic 验证错误。幸运的是，这一次这些问题更容易解决，因为 Workflows 能够很好地处理成员函数。最终，我不得不更加规范地为我的技能创建 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))))

Workflows 的优势

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

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

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

基于事件的发射器-接收器架构取代了直接调用我的 Agent 中的某些方法，例如工具调用处理程序。

如果你拥有更复杂的系统，其中包含多个异步触发的步骤，并且可能会发出多个事件，那么此架构对于干净地管理这些步骤非常有用。

Workflows 的其他好处包括它非常轻量级，不会强迫你接受太多结构（除了使用某些 LlamaIndex 对象），并且其基于事件的架构为直接函数调用提供了一种有用的替代方案 — 尤其是对于复杂的异步应用程序。

纵观这三种方法，每一种都有其优点。

无框架方法最容易实现。因为任何抽象都是由开发人员定义的（即上例中的 SkillMap 对象），所以保持各种类型和对象的连贯性很容易。然而，代码的可读性和可访问性完全取决于个人开发人员，而且很容易看出，如果没有一些强制结构，日益复杂的代理会变得多么混乱。

LangGraph 提供了相当多的结构，这使得 Agent 的定义非常明确。如果一个更广泛的团队正在合作开发一个 Agent，这种结构将提供一种有用的方法来强制执行架构。对于那些不太熟悉该结构的人来说，LangGraph 也可能为 Agent 提供一个很好的起点。然而，这有一个权衡——因为 LangGraph 为你做了很多事情，如果你不完全接受框架，它可能会导致麻烦；代码可能非常干净，但你可能会付出更多调试的代价。

Workflows 介于两者之间。基于事件的架构可能对某些项目非常有用，而对 LlamaIndex 类型的使用要求较少的事实为那些没有在其应用程序中充分使用框架的人提供了更大的灵活性。

最终，核心问题可能归结为“你是否已经在使用 LlamaIndex 或 LangChain 来编排你的应用程序？” LangGraph 和 Workflows 都与各自的底层框架紧密相连，因此每个特定于 Agent 的框架的额外优势可能不会让你单凭优点就切换。

纯代码方法可能始终是一个有吸引力的选择。如果你能够严格记录和执行任何创建的抽象，那么确保外部框架中的任何内容都不会减慢你的速度就很容易了。

当然，“视情况而定”永远不是一个令人满意的答案。这三个问题应该可以帮助你决定在下一个 Agent 项目中使用哪个框架。

你是否已经在项目的重要部分中使用了 LlamaIndex 或 LangChain？

如果是，请先探索该选项。

你是否熟悉常见的 Agent 结构，或者你是否希望有人告诉你应该如何构建 Agent？

如果你属于后者，请尝试 Workflows。如果你确实属于后者，请尝试 LangGraph。

你的 Agent 之前是否已构建过？

该框架的一个好处是，每个框架都有许多教程和示例。可供构建的纯代码代理示例要少得多。

选择 Agent 框架只是众多选择之一，这些选择将影响生成式 AI 系统的生产结果。一如既往，拥有强大的护栏和 LLM 跟踪是值得的 — 并且随着新的 Agent 框架、研究和模型颠覆既定技术，保持敏捷性。