使用LangGraph基于DAG构建自动化任务

今天和大家讲一下如何利用LangGraph构建面向企业应用的智能AI系统。

随着企业逐渐迈向自动化运营，他们需要的不仅仅是回答问题的系统，而是能够思考、规划和行动的系统。这些下一代系统必须能够协调多步骤流程，选择最合适的工具或数据源，检索并推理上下文信息，并在没有持续人为干预的情况下自主执行决策。

1. LangGraph：动态构建知识与多跳检索

当然，首先我们得知道LangGraph的核心功能是允许开发者使用基于图的模型定义和执行智能工作流。

在这个模型中，每个计算或决策都被表示为一个节点，而执行顺序由边决定。每个节点封装了一个独特的步骤，比如信息检索、决策制定、摘要生成或分类，而边则决定了信息从一个步骤到下一个步骤的流动方式。这个结构使得LangGraph在必须遵循程序逻辑和条件分支的企业系统中变得高度模块化和直观。

使用DAG（有向无环图）

DAG，即有向无环图，特别适合于对现实世界业务和推理过程进行建模，原因如下：

1. 顺序至关重要：某些任务必须在其他任务之前完成——例如，在转换数据之前验证数据，或在总结文档之前检索文档。

2. 可以定义条件路径：基于不断演变的状态，图可以分支到不同的方向。

3. 无循环确保有限完成：没有循环保证了流程朝着完成迈进，使DAG适用于工作流自动化。

LangGraph将这个图结构编译成一个可调用的函数。在执行过程中，LangGraph的内部引擎处理节点调用，跟踪状态的流动，并确保节点之间的依赖关系和转换得到尊重。这个模型足够灵活用于原型设计，同时又足够健壮用于生产流水线。

多跳检索：数学视角

在智能系统中，多跳检索指的是从多个来源或步骤中综合信息以达到连贯、准确的答案的过程，特别是在单个文档或事实不足以回答问题时。这在答案分布在政策文件、日志、知识库或API中的企业环境中常常需要。

从数学和算法的角度来看，多跳检索被视为图遍历问题，有几种建模策略：

1. 最短路径算法：包括Dijkstra算法或A*搜索等方法，用于识别两点之间（例如，用户查询和相关数据节点）最有效的路径。

2. 图上的强化学习（RL）：在这种范式中，代理通过接收正确答案的奖励来训练导航知识实体图。随着时间的推移，它们学会了最佳的遍历策略，尤其在开放域问答中非常有用。

3. 马尔可夫决策过程（MDP）：图中的每一步或跳跃被建模为一个决策点，具有相关的概率和预期结果。代理选择一条最大化累积信息增益的路径，使MDP成为不确定性感知检索系统的理想选择。

在LangGraph中，多跳推理不是抽象的，而是嵌入在图的结构中。例如，一个代理可能首先识别一个实体，然后查找其属性，最后检索相关文档，每个步骤都是执行图中的一个不同节点。代理不仅获取数据，还根据刚刚学到的内容决定下一步要获取什么。

2. 编码文档、主题和上下文之间的互连

在企业环境中，上下文并不局限于单一文档，而是分布在多个数据源、部门和格式中。用户查询可能引用政策文件中定义的主题、知识库中详细说明的内容以及合规检查表中强制执行的内容。为了让智能代理有效推理，它们必须能够连接和遍历这些相互关联的概念。LangChain和LangGraph通过启用符号和语义知识表示的集成提供了处理这种复杂性的机制。

第一种方法是符号建模，它明确地表示实体及其关系。这通常使用Neo4j等图形数据库或RDF等标准实现。符号表示在需要清晰、可检查链接的情况下是理想的——例如，理解文档引用的是哪项法规，或者客户投诉如何与特定产品线相关。例如，要检索所有与GDPR相关的文档，可以使用通过Neo4j集成的Cypher查询：

from langchain.graphs import Neo4jGraph
graph = Neo4jGraph(
    url="bolt://localhost:7687", 
    username="neo4j", 
    password="your_password"
)
query = """
MATCH (doc:Document)-[:REFERS_TO]->(topic:Topic) 
WHERE topic.name = 'GDPR' 
RETURN doc.title
"""
results = graph.query(query)

这些结果随后可以输入到LangGraph节点中，该节点更新共享状态，允许下游代理使用上下文感知信息操作，例如根据文档相关性或优先级定制响应。

第二种方法是语义建模，其中文档、问题和实体被嵌入到高维向量空间中。这是Pinecone和OpenAI Embeddings等工具大显身手的地方。语义搜索擅长查找概念上相似的内容，即使确切的关键词或结构不同。例如：

1. 可以检索与查询语义相关的内容，即使术语有所不同。

2. 支持快速近似最近邻（ANN）查找，可扩展到数百万个向量。

3. 支持模糊、概念驱动的推理，其中确切的结构并不总是可用。

然而，这两种方法都有优点和局限性。这就是为什么当今许多企业级系统采用混合策略：

1. 使用Pinecone（语义检索）提高性能、语言灵活性和内容发现能力。

2. 使用Neo4j或内存内图（符号）确保结构、可追溯性和精确链接建模。

这种混合模型赋予智能系统平衡精度和相关性的能力，为代理提供了一张关系图和一个语义指南针，帮助它们在松散结构化或快速变化的数据环境中导航。

3. 实现多步推理与智能行为

LangGraph最具定义性的能力之一是支持多步推理，这对于构建模仿人类解决复杂问题的智能工作流来说是必需的。在LangGraph中，执行图中的每个节点不仅仅执行静态任务，而是代表一个独特的推理阶段。这些节点设计为在基于Pydantic的共享状态下操作，它们可以检查、修改并向前传递。这种有状态的设计允许节点了解过去的计算，从而在连续步骤中实现复杂的逻辑。

在工作流的每个阶段，一个节点接收到系统的当前状态——这可能包括输入、过去的结果、检索到的文档或之前的决策——并使用这些来决定接下来做什么。一旦节点完成其动作，它返回一个更新后的状态，然后根据定义的DAG流向下一个节点。

这种架构为智能行为奠定了基础，其中系统不仅仅遵循线性路径，而是在每一步做出决策。以下是智能行为在实践中真正意味着什么：

1. 系统对输入进行推理：它不仅分析即时提示或指令，还分析上下文变量，如先前检索到的数据、外部工具结果和内部内存。

2. 它选择使用哪种工具或路径：基于不断演变的状态，代理可以动态决定是否执行数据库查找、启动网络搜索、总结检索到的内容或提出后续问题。

3. 它随着上下文演变而适应：如果某个工具失败，或者输入在执行过程中发生变化（例如，由于用户更正或新数据），代理可以调整——跳过不必要的步骤或重新访问早期步骤。

这种灵活的、基于状态的逻辑赋予智能系统比传统流水线架构明显的优势。

4. 使用LangChain代理进行工具集成

LangChain最强大的功能之一是其对代理的支持，这是一种智能组件，能够解释用户输入并自主决定使用哪些工具、以何种顺序、在何种条件下使用。与静态链或硬编码脚本不同，LangChain代理支持动态决策，允许系统根据大型语言模型（LLM）的实时推理按需选择工具。这在企业场景中特别有价值，因为查询可能是广泛的、多方面的或模糊的。

LangChain提供了一个标准化的代理接口，简化了工具的集成，无论是网络搜索API、CRM数据库、文档检索器还是自定义企业功能。每个工具被定义为一个带有名称和描述的可调用Python函数，代理根据当前提示或任务评估其相关性。

定义工具

要在LangChain中定义工具，您将函数包装在一个Tool对象中。这个函数可以执行从查找产品ID到调用第三方服务的任何操作。以下是一个简单搜索工具的示例：

from langchain.tools import Tool
search_tool = Tool(
    name="SearchTool",
    func=lambda query: f"Search results for: {query}",
    description="Useful for web search queries"
)

此工具稍后可以注册到代理中，代理将其视为可用操作之一，当决定如何响应用户查询时会考虑它。

使用工具与代理

一旦您的工具定义好了，您可以初始化一个使用LLM确定何时以及如何使用这些工具的LangChain代理。以下是一个使用OpenAI GPT模型的最小示例：

from langchain.Agents import initialize_agent, AgentType
from langchain.chat_models import ChatOpenAI
llm = ChatOpenAI()
agent = initialize_agent(
    tools=[search_tool],
    llm=llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
)
response = agent.run("Find the latest developments in quantum computing")
print(response)

在这个例子中，代理解释用户的查询并选择SearchTool作为最合适的工具来调用。LLM不仅输出响应，它还分析输入，选择工具，调用工具，然后根据工具的结果合成最终答案。

通过LangChain代理进行工具集成不仅是技术便利，更是企业级智能系统的战略推动者。与LangGraph结合，这些代理可以作为智能节点放置在DAG中，允许AI工作流的每个阶段以工具增强的自主性操作。

5. 创建计划、检索和执行的自主AI系统

我们目前为止所涵盖的所有内容，从基于图的结构化和多跳检索到工具集成和智能推理，都汇聚到这一最终层面；创建完全自主的AI系统，这些系统能够独立计划步骤、检索相关信息并执行任务。

在LangGraph中，这种自主性是通过结合结构化的状态管理、工具增强的推理和动态决策逻辑实现的，所有这些都连接到基于DAG的执行图中。这种架构允许您设计模仿人类工作流的AI系统，但以机器速度运行，并且具有更高的可扩展性和一致性：

1. 使用Pydantic模型定义共享状态：这充当一种存储代理在执行期间收集或转换的数据的记忆结构——例如原始查询、工具输出、中间决策和最终结果。

2. 将图形节点创建为单独的Python函数：每个节点应接受当前状态，执行任务（例如，工具调用、LLM调用、推理步骤），然后返回更新后的状态版本。

3. 使用边连接这些节点以定义执行路径：边指定哪个节点跟随另一个节点，允许线性流、分支条件或并行处理，具体取决于您的用例。

4. 直接将工具集成到节点逻辑中或将它们包裹在LangChain代理中：这些工具可以包括搜索API、文档检索器、内部数据库或外部SaaS端点。

一旦这个图被编译，LangGraph就会接管编排，也就是按顺序执行节点，将状态向前路由，并根据状态值或代理输出进行动态决策。现在让我们看看三个现实的、简化的代码示例，展示这些系统在实践中的工作原理，将LangChain工具、OpenAI函数、Pinecone矢量检索和LangGraph DAG联系在一起。

OpenAI函数调用（结构化输出）

这是如何使用通过LangChain的OpenAI函数调用从非结构化文本中提取结构化数据——在诸如表格填写、实体提取或解析报告等场景中极为有用。

# 使用Pydantic定义数据模式——这告诉LLM我们期望的结构
from langchain.pydantic_v1 import BaseModel, Field
class Person(BaseModel):
    name: str = Field(..., description="The person's name")
    age: int = Field(..., description="The person's age")

在这里，Person是一个类型化的数据模型。Field描述有助于引导LLM产生正确的格式（在底层JSON可解析）。

# 导入LLM并创建使用函数调用的链
from langchain.chains.openai_functions import create_structured_output_chain
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
# 初始化LLM——在这种情况下，GPT-4具有确定性输出
llm = ChatOpenAI(model="gpt-4", temperature=0)

零温度模型确保一致的输出——对于结构化提取至关重要。

# 构造围绕用户输入的提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a world-class algorithm for extracting structured data."),
    ("human", "Use the given format to extract data: {input}"),
])

系统消息告诉LLM它应该做什么，而人类提示传递输入。这是灵活且可重用的。

# 使用模式、LLM和提示创建结构化输出链
chain = create_structured_output_chain(Person, llm, prompt)
# 使用自然语言输入运行链
result = chain.run("Tom is 42 years old")
print(result)  # → Person(name="Tom", age=42)

一个经过验证的Person对象，其中填入了姓名和年龄字段——非常适合下游使用，如存储在数据库中或传递到另一个管道中。

使用Pydantic进行DAG执行的LangGraph

现在让我们构建一个简单的LangGraph DAG，展示智能逻辑是如何通过以下方式结构化的：

• 共享状态（AgentState）

• 修改该状态的节点

• 连接推理步骤的图流

# 定义共享状态模型
from langgraph.graph import StateGraph, END
from langgraph.pydantic import BaseModel
class AgentState(BaseModel):
    input_query: str
    retrieved_docs: list[str] = []
    answer: str = ""

AgentState对象保存了代理在执行期间知道或决定的一切——查询、找到的任何文档和最终答案。

# 节点1：模拟检索
def retrieve_docs(state):
    state.retrieved_docs = ["DocA", "DocB"]  # 模拟检索
    return state

您可以在这里插入一个真实的Pinecone调用——这只是占位符。

# 节点2：从检索到的文档中生成答案
def generate_answer(state):
    state.answer = f"Answer based on: {state.retrieved_docs}"
    return state

通常在这里调用您的LLM，可能参考检索到的文档作为上下文。

# 构建LangGraph工作流
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_docs)
workflow.add_node("respond", generate_answer)
workflow.add_edge("retrieve", "respond")
workflow.add_edge("respond", END)

这定义了一个2节点DAG，具有清晰的流程：“检索”→“响应”→END。

# 编译并运行工作流
agent_app = workflow.compile()
result = agent_app.invoke({"input_query": "What is LangGraph?"})
print(result)

这就是如何构建一个完全可执行的LangGraph应用程序，它可以扩展到包含分支、工具使用、条件或错误处理，形成真实世界智能系统的基础。

正如我们在整个指南中探讨的那样，为企业应用部署现代智能AI系统要求的不仅仅是简单地将一些提示串联起来。它需要精心设计有状态、模块化和决策驱动的工作流，以反映实际团队的操作方式。LangGraph在实现这种智能编排方面发挥了核心作用。作为一种基于DAG的执行框架，LangGraph为以模块化和透明的方式协调复杂工作流提供了结构基础。当与LangChain的强大代理接口、OpenAI的函数调用能力和像Pinecone这样的语义向量搜索系统结合时，开发人员可以构建既智能又可靠的稳健且可扩展的AI系统。

简而言之，这种架构将范式从基本聊天界面转变为真正自主的企业级数字代理，可以在现实商业逻辑的范围内搜索、推理、决策和行动。这不仅仅是让AI更有用，而是让它像值得信赖的队友一样工作。