用 Agentic 框架构建 AI 工作流
随着 GPT-5、Gemini 2.5 Pro 等强大 AI 模型的涌现，旨在高效利用这些模型的 Agentic 框架也日益增多。这类框架通过抽象化诸多复杂环节，极大地简化了与 AI 模型的协作，例如处理工具调用、管理智能体状态以及集成人工反馈循环。
本文将深入探讨其中一个可用的 Agentic AI 框架：LangGraph。我们将用它来开发一个简单的智能体应用，通过具体步骤展示此类框架的优势，并在文末讨论使用 LangGraph 及其他类似框架的优缺点。
市面上有多种 Agentic 框架可供选择，例如：
* LangChain
* LlamaIndex
* CrewAI
为什么需要 Agentic 框架？
尽管市面上存在许多旨在简化应用开发的库，但它们有时反而会使代码变得晦涩、影响生产环境性能，并增加调试难度。
关键在于找到那些能通过抽象样板代码来真正简化应用的库。这一原则在创业领域常被概括为：专注于解决你的核心问题，而将那些已被解决的问题交给成熟的工具去处理。
Agentic 框架的价值在于它能够抽象掉开发者通常不愿处理的复杂性：
* 状态管理：不仅管理对话历史，还包括在执行 RAG 等任务时收集的所有相关信息。
* 工具使用：开发者无需编写调用工具的具体逻辑，只需定义好工具，框架便能负责如何调用，尤其擅长处理并行与异步的工具调用。
因此，使用 Agentic 框架可以剥离大量底层细节，让开发者能够将精力集中于产品的核心逻辑。
LangGraph 基础
LangGraph 的核心思想是基于“图”来构建工作流。在每次处理请求时，系统都会执行这个图。图中主要包含三个要素：
* 状态：保存在内存中的当前信息。
* 节点：通常是执行特定操作的单元，例如调用 LLM 判断意图或调用工具执行任务。
* 边：定义节点之间的流转逻辑，通常基于条件判断来决定下一步执行哪个节点。
这些概念均源自基础的图论。
实现一个工作流
我们将通过一个简单的文档处理应用来实践 LangGraph。该应用允许用户执行以下操作：
* 创建带文本的文档
* 删除文档
* 在文档中搜索
为此，我们将构建一个包含两个主要步骤的工作流：
1. 识别用户意图：判断用户是想创建、删除还是搜索文档。
2. 路由执行：根据识别的意图，将请求路由到对应的处理流程。
虽然也可以直接让智能体自由调用所有工具，但先进行意图分类的路由设计，能为后续根据意图执行更复杂的操作序列提供清晰的架构。
加载依赖与 LLM
首先，加载必要的库并初始化 LLM。本例中使用 AWS Bedrock 的 Claude 模型，你也可以替换为其他服务商。
from typing_extensions import TypedDict, Literal
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, Field
from IPython.display import display, Image
from dotenv import load_dotenv
import os
load_dotenv()
aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID") or ""
aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY") or ""
os.environ["AWS_ACCESS_KEY_ID"] = aws_access_key_id
os.environ["AWS_SECRET_ACCESS_KEY"] = aws_secret_access_key
llm = ChatBedrockConverse(
model_id="us.anthropic.claude-3-5-haiku-20241022-v1:0",
region_name="us-east-1",
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
)
# 使用字典模拟文档数据库，生产环境应替换为真实数据库
document_database: dict[str, str] = {}
定义图结构
接下来定义图。首先创建一个路由器，用于将用户输入分类为三种意图之一：add_document、delete_document 或 ask_document。
# 定义状态结构
class State(TypedDict):
input: str
decision: str | None
output: str | None
# 为路由逻辑定义结构化输出模式
class Route(BaseModel):
step: Literal["add_document", "delete_document", "ask_document"] = Field(
description="路由流程中的下一步"
)
# 增强 LLM，使其支持结构化输出以用于路由
router = llm.with_structured_output(Route)
def llm_call_router(state: State):
"""将用户输入路由到适当的节点"""
decision = router.invoke(
[
SystemMessage(
content="""将用户输入路由到以下三种意图之一：
- 'add_document'
- 'delete_document'
- 'ask_document'
你只需返回意图，无需其他文本。
"""
),
HumanMessage(content=state["input"]),
]
)
return {"decision": decision.step}
# 条件边函数，根据决策路由到对应节点
def route_decision(state: State):
if state["decision"] == "add_document":
return "add_document_to_database_tool"
elif state["decision"] == "delete_document":
return "delete_document_from_database_tool"
elif state["decision"] == "ask_document":
return "ask_document_tool"
这里定义了状态 State 来存储用户输入和路由决策。通过强制 LLM 进行结构化输出，我们确保了模型只会返回三种预定义的意图之一，从而实现了可靠的路由逻辑。
接着，我们定义本例中使用的三个工具，每个意图对应一个工具。
# 节点定义
def add_document_to_database_tool(state: State):
"""向数据库添加文档。根据用户查询，提取文档的文件名和内容。若未提供，则不添加。"""
user_query = state["input"]
# 从用户查询中提取文件名
filename_prompt = f"根据以下用户查询，提取文档的文件名：{user_query}。只返回文件名，不要返回其他文本。"
output = llm.invoke(filename_prompt)
filename = output.content
# 从用户查询中提取内容
content_prompt = f"根据以下用户查询，提取文档的内容：{user_query}。只返回内容，不要返回其他文本。"
output = llm.invoke(content_prompt)
content = output.content
# 将文档添加到数据库
document_database[filename] = content
return {"output": f"文档 {filename} 已添加到数据库"}
def delete_document_from_database_tool(state: State):
"""从数据库删除文档。根据用户查询，提取要删除的文档的文件名。若未提供，则不删除。"""
user_query = state["input"]
# 从用户查询中提取文件名
filename_prompt = f"根据以下用户查询，提取要删除的文档的文件名：{user_query}。只返回文件名，不要返回其他文本。"
output = llm.invoke(filename_prompt)
filename = output.content
# 如果文档存在则删除，否则返回失败信息
if filename not in document_database:
return {"output": f"数据库中未找到文档 {filename}"}
document_database.pop(filename)
return {"output": f"文档 {filename} 已从数据库删除"}
def ask_document_tool(state: State):
"""询问文档相关问题。根据用户查询，提取文档的文件名和问题。若未提供，则不提问。"""
user_query = state["input"]
# 从用户查询中提取文件名
filename_prompt = f"根据以下用户查询，提取要提问的文档的文件名：{user_query}。只返回文件名，不要返回其他文本。"
output = llm.invoke(filename_prompt)
filename = output.content
# 从用户查询中提取问题
question_prompt = f"根据以下用户查询，提取要问文档的问题：{user_query}。只返回问题，不要返回其他文本。"
output = llm.invoke(question_prompt)
question = output.content
# 对文档进行提问
if filename not in document_database:
return {"output": f"数据库中未找到文档 {filename}"}
result = llm.invoke(f"文档：{document_database[filename]}nn问题：{question}")
return {"output": f"文档查询结果：{result.content}"}
最后，我们通过添加节点和边来构建图：
# 构建工作流
router_builder = StateGraph(State)
# 添加节点
router_builder.add_node("add_document_to_database_tool", add_document_to_database_tool)
router_builder.add_node("delete_document_from_database_tool", delete_document_from_database_tool)
router_builder.add_node("ask_document_tool", ask_document_tool)
router_builder.add_node("llm_call_router", llm_call_router)
# 添加边以连接节点
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
"llm_call_router",
route_decision,
{ # route_decision 返回的名称 : 要访问的下一个节点名称
"add_document_to_database_tool": "add_document_to_database_tool",
"delete_document_from_database_tool": "delete_document_from_database_tool",
"ask_document_tool": "ask_document_tool",
},
)
router_builder.add_edge("add_document_to_database_tool", END)
router_builder.add_edge("delete_document_from_database_tool", END)
router_builder.add_edge("ask_document_tool", END)
# 编译工作流
memory = InMemorySaver()
router_workflow = router_builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "1"}}
# 可视化工作流
display(Image(router_workflow.get_graph().draw_mermaid_png()))
最后的显示函数会渲染出如下图所示的图结构：
现在，我们可以按不同意图来运行这个工作流了。
添加文档：
user_input = "Add the document 'test.txt' with content 'This is a test document' to the database"
state = router_workflow.invoke({"input": user_input}, config)
print(state["output"])
# -> Document test.txt added to database
查询文档：
user_input = "Give me a summary of the document 'test.txt'"
state = router_workflow.invoke({"input": user_input}, config)
print(state["output"])
# -> A brief, generic test document with a simple descriptive sentence.
删除文档：
user_input = "Delete the document 'test.txt' from the database"
state = router_workflow.invoke({"input": user_input}, config)
print(state["output"])
# -> Document test.txt deleted from database
可以看到，工作流在不同的路由路径下都能正确运行。你可以根据需要轻松地增加更多意图，或为每个意图添加更多节点，从而构建更复杂的工作流。
更强的 Agentic 用例
“Agentic Workflows” 与完全的 “Agentic Applications” 之间的区别有时会令人困惑。为了区分这两个术语，这里引用 Anthropic 在《Building effective agents》一文中的观点：
工作流 是通过预定义的代码路径来编排 LLM 和工具的系统；而 智能体 是由 LLM 动态地指挥其自身流程与工具使用、并持续控制任务完成方式的系统。
大多数使用 LLM 解决的问题更适合采用工作流模式，因为多数问题（根据经验）是预定义的，并且应该遵循一套预设的防护栏。例如，在上述添加/删除/查询文档的场景中，最佳实践就是先定义意图分类器，并基于不同意图设定好后续的固定流程。
但在某些场景下，你可能需要构建更具自主性的智能体应用。例如，一个能够在你代码库中搜索、在线查阅最新文档并直接修改代码的编程助手。这类应用的潜在场景极为多样，很难预先定义固定的工作流。
若想构建具备更高自主性的智能体系统，可以深入探索其核心概念。
LangGraph 的优缺点
优点
我认为 LangGraph 主要有以下三个优点：
- 上手简单：安装和快速启动非常便捷。参考官方文档，或利用 AI 助手（如 Cursor）基于文档实现特定工作流，都能轻松开始。
- 开源：其代码完全开源。这意味着无论背后的公司如何变化，你都可以确保项目代码的长期可运行性，这对于生产环境至关重要。
- 代码简洁：它显著简化了大量样板代码，并抽象掉了许多原本需要手动处理的复杂逻辑，让开发更聚焦于业务本身。
缺点
在实际使用中，我也发现了一些不足之处：
- 仍有一定样板代码：在实现自定义工作流时，虽然比从零开始写代码量少，但仍需要编写一定量的配置和结构代码，以实现一个相对简单的流程。这部分源于 LangGraph 定位为一个比某些高层框架（如 LangChain）更底层的工具，旨在提供灵活性的同时避免过度抽象。
- 特有的集成问题：如同许多第三方库，集成 LangGraph 时可能会遇到其特有的错误。例如，在尝试可视化工作流图时，我遇到了与
_draw_mermaid_png函数相关的问题。使用外部库总是在“获得的便利抽象”和“可能遇到的特定 Bug”之间进行权衡。
总结
总体而言，我认为 LangGraph 是构建智能体系统的一个非常有用的工具。通过意图分类来路由到不同处理流程的方式，让我能够相对轻松地搭建所需的工作流。它在“避免过度抽象以保持代码透明和易于调试”与“封装不必要的复杂性”之间取得了良好的平衡。采用此类智能体框架有利有弊，而判断其是否适合你的最佳方式，就是亲自尝试实现几个简单的工作流。
原文地址：https://pub.towardsai.net/how-to-build-effective-agentic-systems-with-langgraph-e5433d7aa153
关注“鲸栖”小程序，掌握最新AI资讯
本文由鲸栖原创发布，未经许可，请勿转载。转载请注明出处：http://www.itsolotime.com/archives/13403