1. LangGraph进阶：构建ReACT架构的智能Agent

1.1 引言

在人工智能和大语言模型（LLM）快速发展的今天，如何构建高效、灵活的智能Agent成为了一个热门话题。LangGraph作为一个强大的工具，为我们提供了一种新的方式来实现复杂的AI工作流，特别是在构建ReACT（Reasoning and Acting）架构的智能Agent方面表现出色。本文将深入探讨如何使用LangGraph来实现ReACT架构，并提供详细的代码示例和解释。

1.2 LangGraph的基本概念

LangGraph是一个用于构建基于LLM的应用程序的Python框架。它的核心理念是将复杂的AI工作流程表示为一个状态图，其中包含节点、边和数据状态。这种方法使得我们可以更直观地设计和实现智能Agent的行为逻辑。

在LangGraph中，我们可以使用基础组件（节点、边、数据状态）来构建Agent，这也是LangGraph自由度高的一个重要优点。同时，LangGraph还提供了一些预构建的代理，如ReACT智能体和工具调用智能体，使得我们可以更快速地创建智能Agent。

1.3 ReACT架构简介

ReACT（Reasoning and Acting）是一种智能Agent架构，它结合了推理和行动的能力。在ReACT架构中，Agent通过不断地思考（Reasoning）、采取行动（Acting）和观察结果（Observing）来解决问题。这种方法使得Agent能够更灵活地应对复杂的任务，并且可以利用外部工具来增强其能力。

使用LangGraph实现ReACT架构
现在，让我们通过一个具体的例子来看看如何使用LangGraph实现ReACT架构的智能Agent。

1.3.1 环境设置

首先，我们需要导入必要的库和模块：

1
2
3
4
5
6
7
8
9
10
11
12
import dotenv
from langchain_community.tools import GoogleSerperRun
from langchain_community.tools.openai_dalle_image_generation import OpenAIDALLEImageGenerationTool
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_community.utilities.dalle_image_generator import DallEAPIWrapper
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.prebuilt.chat_agent_executor import create_react_agent


dotenv.load_dotenv()

1.3.2 定义工具和参数模式

接下来，我们定义了两个工具：Google搜索和DALL-E图像生成。同时，我们也为这些工具定义了参数模式：

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
class GoogleSerperArgsSchema(BaseModel):
    query: str = Field(description="执行谷歌搜索的查询语句")


class DallEArgsSchema(BaseModel):
    query: str = Field(description="输入应该是生成图像的文本提示(prompt)")

google_serper = GoogleSerperRun(
    name="google_serper",
    description=(
        "一个低成本的谷歌搜索API。"
        "当你需要回答有关时事的问题时，可以调用该工具。"
        "该工具的输入是搜索查询语句。"
    ),
    args_schema=GoogleSerperArgsSchema,
    api_wrapper=GoogleSerperAPIWrapper(),
)

dalle = OpenAIDALLEImageGenerationTool(
    name="openai_dalle",
    api_wrapper=DallEAPIWrapper(model="dall-e-3"),
    args_schema=DallEArgsSchema,
)

tools = [google_serper, dalle]

1.3.3 创建语言模型

我们使用OpenAI的GPT-4模型作为我们的大语言模型：

1
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

1.3.4 使用预构建函数创建ReACT智能体

LangGraph提供了预构建的函数来创建ReACT智能体，使用起来非常简单：

1
2
3
4
agent = create_react_agent(
    model=model,
    tools=tools
)

1.3.5 调用智能体并输出内容

最后，我们可以调用我们创建的智能体，并打印输出结果：

1
print(agent.invoke({"messages": [("human", "帮我绘制一幅鲨鱼在天上飞的图片")]}))

1.3.6 运行结果分析

当我们运行这段代码时，智能体会首先理解任务需求，然后决定使用DALL-E工具来生成图像。它会生成一个详细的图像描述，然后调用DALL-E API来创建图像。最后，它会返回生成的图像URL，并提供一个简短的描述。

输出内容可能如下所示：

1
2
3
4
5
{'messages': [HumanMessage(content='帮我绘制一幅鲨鱼在天上飞的图片'), 
              AIMessage(content='', additional_kwargs={'tool_calls': [...]}), 
              ToolMessage(content='https://dalleproduse.blob.core.windows.net/...'), 
              AIMessage(content='这是您请求的画面：一只鲨鱼在天上飞翔的图片。您可以点击下面的链接查看这幅图。\n\n![鲨鱼在天上飞](https://dalleproduse.blob.core.windows.net/...)')]
}

1.4 总结

通过这个例子，我们可以看到LangGraph如何简化了ReACT架构智能Agent的构建过程。它提供了高级抽象和预构建组件，使得我们可以快速实现复杂的AI工作流。同时，LangGraph的灵活性也允许我们根据需要自定义和扩展Agent的功能。

需要注意的是，LangGraph仍在快速发展中。例如，当前版本（截至文章撰写时）的预构建ReACT智能体是基于函数调用实现的，并且在0.3.0版本中可能会被移除。因此，在使用LangGraph时，建议关注其最新的文档和更新。

尽管如此，LangGraph的核心设计理念和封装思路仍然非常值得学习和借鉴。随着AI技术的不断进步，我们可以期待LangGraph在未来会提供更多强大和易用的功能，进一步简化智能Agent的开发过程。

2. LangGraph高级技巧：图结构应用程序删除消息的使用技巧

2.1 消息删除的重要性

在图结构应用中，消息的累积可能会导致性能问题和不必要的复杂性。因此，适时删除不再需要的消息是很有必要的。LangGraph提供了几种方法来实现这一目标。

2.2 使用delete_messages函数

LangGraph提供了delete_messages函数，它可以根据指定的条件删除消息。这个函数可以在图的节点中使用，特别是在处理完某些消息后。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
from langgraph.prebuilt import ToolMessage

def process_and_delete(state):
    # 处理消息的逻辑
    # ...
    
    # 删除已处理的消息
    state = delete_messages(state, lambda x: isinstance(x, ToolMessage))
    return state

# 在图结构中使用
graph = Graph()
graph.add_node("process_and_delete", process_and_delete)
# ...

2.3 消息过滤技巧

除了直接删除消息，我们还可以使用过滤技巧来控制消息的流动：

2.3.1 使用条件边

1
2
3
4
5
graph.add_conditional_edge(
    "node_a",
    "node_b",
    condition=lambda x: not isinstance(x, ToolMessage)
)

2.3.2 在节点函数中进行过滤

1
2
3
4
5
def filter_messages(state):
    filtered_messages = [msg for msg in state['messages'] if not isinstance(msg, ToolMessage)]
    return {"messages": filtered_messages}

graph.add_node("filter", filter_messages)

2.4 消息修剪策略

在长时间运行的应用中，可能需要定期修剪消息以保持性能：

2.4.1 基于时间的修剪

1
2
3
4
5
6
7
8
9
10
11
from datetime import datetime, timedelta

def prune_old_messages(state):
    current_time = datetime.now()
    recent_messages = [
        msg for msg in state['messages']
        if current_time - msg.timestamp < timedelta(hours=1)
    ]
    return {"messages": recent_messages}

graph.add_node("prune", prune_old_messages)

2.4.2 基于数量的修剪

1
2
3
4
def keep_latest_messages(state, max_messages=50):
    return {"messages": state['messages'][-max_messages:]}

graph.add_node("prune", lambda s: keep_latest_messages(s, max_messages=100))

2.5 实际应用案例：智能客服系统

考虑一个智能客服系统，它需要处理用户查询，但同时要保持对话的简洁性：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from langgraph.prebuilt import ToolMessage, HumanMessage

def process_query(state):
    # 处理用户查询
    # ...
    return state

def summarize_and_prune(state):
    # 总结对话
    summary = summarize_conversation(state['messages'])
    
    # 保留最新的人类消息和总结
    new_messages = [msg for msg in state['messages'] if isinstance(msg, HumanMessage)][-3:]
    new_messages.append(ToolMessage(content=summary))
    
    return {"messages": new_messages}

graph = Graph()
graph.add_node("process_query", process_query)
graph.add_node("summarize_and_prune", summarize_and_prune)
graph.add_edge("process_query", "summarize_and_prune")
graph.add_edge("summarize_and_prune", "process_query")

2.6 总结

通过合理使用LangGraph提供的消息删除和过滤功能，我们可以有效地管理图结构应用中的消息流。这不仅可以提高应用的性能，还能使对话更加清晰和有针对性。在实际应用中，根据具体需求选择适当的消息管理策略至关重要。

3. LangGraph中的检查点与人机交互

3.1 LangGraph的检查点机制

检查点机制是LangGraph中一个强大的功能，它允许我们在图执行的特定点暂停处理，保存状态，并在需要时恢复。

3.1.1 检查点的基本概念

检查点本质上是图执行过程中的一个快照，包含了当前的状态信息。这对于长时间运行的任务、需要人工干预的流程，或者需要断点续传的应用特别有用。

3.1.2 创建检查点

在LangGraph中，我们可以使用create_checkpoint函数来创建检查点：

1
2
3
4
5
6
7
8
9
10
11
from langgraph.checkpoint import create_checkpoint

def process_with_checkpoint(state):
    # 处理逻辑
    # ...
    
    # 创建检查点
    checkpoint = create_checkpoint(state)
    return {"checkpoint": checkpoint, "state": state}

graph.add_node("process", process_with_checkpoint)

3.1.3 恢复检查点

使用load_checkpoint函数可以恢复之前保存的检查点：

1
2
3
4
5
6
7
8
9
from langgraph.checkpoint import load_checkpoint

def resume_from_checkpoint(checkpoint):
    state = load_checkpoint(checkpoint)
    # 继续处理
    # ...
    return state

graph.add_node("resume", resume_from_checkpoint)

3.2 实现人在环路(Human-in-the-loop)交互

人在环路交互是指在AI系统的执行过程中，允许人类参与并提供输入或做出决策。LangGraph提供了灵活的机制来实现这种交互。

3.2.1 使用回调函数实现人机交互

我们可以在图的节点中定义回调函数，用于获取人类输入：

1
2
3
4
5
6
7
8
9
10
11
12
def human_input_node(state):
    # 显示当前状态给用户
    print("Current state:", state)
    
    # 获取用户输入
    user_input = input("Please provide your input: ")
    
    # 更新状态
    state['user_input'] = user_input
    return state

graph.add_node("human_input", human_input_node)

3.2.2 条件分支实现人机交互

我们可以使用条件分支来决定是否需要人类干预：

1
2
3
4
5
6
7
8
9
10
11
12
13
def check_confidence(state):
    if state['confidence'] < 0.8:
        return "human_input"
    else:
        return "auto_process"

graph.add_conditional_edges(
    "check_confidence",
    {
        "human_input": lambda s: s['confidence'] < 0.8,
        "auto_process": lambda s: s['confidence'] >= 0.8
    }
)

3.3 实际应用案例：智能客服系统升级版

让我们将检查点机制和人机交互结合到之前的智能客服系统中：

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from langgraph.prebuilt import ToolMessage, HumanMessage
from langgraph.checkpoint import create_checkpoint, load_checkpoint

def process_query(state):
    # 处理用户查询
    # ...
    state['confidence'] = calculate_confidence(state)
    return state

def human_intervention(state):
    print("Current conversation:", state['messages'])
    human_response = input("Please provide assistance: ")
    state['messages'].append(HumanMessage(content=human_response))
    return state

def summarize_and_prune(state):
    # 总结对话
    summary = summarize_conversation(state['messages'])
    
    # 保留最新的消息和总结
    new_messages = state['messages'][-5:]
    new_messages.append(ToolMessage(content=summary))
    
    state['messages'] = new_messages
    
    # 创建检查点
    checkpoint = create_checkpoint(state)
    state['checkpoint'] = checkpoint
    
    return state

graph = Graph()
graph.add_node("process_query", process_query)
graph.add_node("human_intervention", human_intervention)
graph.add_node("summarize_and_prune", summarize_and_prune)

graph.add_conditional_edges(
    "process_query",
    {
        "human_intervention": lambda s: s['confidence'] < 0.8,
        "summarize_and_prune": lambda s: s['confidence'] >= 0.8
    }
)
graph.add_edge("human_intervention", "summarize_and_prune")
graph.add_edge("summarize_and_prune", "process_query")

在这个升级版的智能客服系统中，我们引入了以下改进：

  • 根据置信度决定是否需要人工干预。
  • 在每次总结和修剪后创建检查点，以便在需要时恢复状态。
  • 人工干预节点允许人类直接参与对话。

3.4 总结

LangGraph的检查点机制和人机交互功能为构建复杂、可靠的AI系统提供了强大的工具。通过合理使用这些功能，我们可以创建更加智能、灵活且可控的应用程序。检查点允许我们在长时间运行的任务中保存和恢复状态，而人机交互则为AI决策过程引入了人类的判断和专业知识。在实际应用中，这些功能的结合可以显著提高系统的性能和可靠性。

4. 使用LangGraph构建复杂AI工作流：子图架构详解

4.1 子图架构概述

子图（Subgraph）是LangGraph中一个强大的特性，它允许我们将复杂的工作流程分解成更小、更易管理的组件。通过子图，我们可以实现模块化设计，提高代码的可重用性和可维护性。

4.1.1 子图的基本概念

子图本质上是一个完整的图结构，可以作为更大图结构中的一个节点使用。它具有以下特点：

1
2
3
4
5
6
7
8
9
10
11
from langgraph.graph import SubGraph, Graph

# 创建子图
class MarketingSubGraph(SubGraph):
    def __init__(self):
        super().__init__()
        
    def build(self) -> Graph:
        graph = Graph()
        # 定义子图内部结构
        return graph

4.1.2 子图的优势

  • 模块化：将复杂逻辑封装在独立的子图中
  • 可重用性：子图可以在不同的主图中重复使用
  • 可维护性：更容易进行测试和调试
  • 扩展性：便于添加新功能和修改现有功能

4.2 子图的实现方法

4.2.1 创建基本子图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
class ContentGenerationSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        # 添加内容生成节点
        graph.add_node("generate_content", self.generate_content)
        graph.add_node("review_content", self.review_content)
        
        # 添加边
        graph.add_edge("generate_content", "review_content")
        
        return graph
    
    def generate_content(self, state):
        # 内容生成逻辑
        return state
    
    def review_content(self, state):
        # 内容审查逻辑
        return state

4.2.2 子图的状态管理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class AnalyticsSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def process_analytics(state):
            # 确保状态包含必要的键
            if 'metrics' not in state:
                state['metrics'] = {}
            
            # 处理分析数据
            state['metrics']['engagement'] = calculate_engagement(state)
            return state
            
        graph.add_node("analytics", process_analytics)
        return graph

4.3 子图的组合与交互

4.3.1 在主图中使用子图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def create_marketing_workflow():
    main_graph = Graph()
    
    # 实例化子图
    content_graph = ContentGenerationSubGraph()
    analytics_graph = AnalyticsSubGraph()
    
    # 将子图添加到主图
    main_graph.add_node("content", content_graph)
    main_graph.add_node("analytics", analytics_graph)
    
    # 连接子图
    main_graph.add_edge("content", "analytics")
    
    return main_graph

4.3.2 子图间的数据传递

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class DataProcessingSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def prepare_data(state):
            # 准备数据供其他子图使用
            state['processed_data'] = {
                'content_type': state['raw_data']['type'],
                'metrics': state['raw_data']['metrics'],
                'timestamp': datetime.now()
            }
            return state
            
        graph.add_node("prepare", prepare_data)
        return graph

4.4 实际案例：营销智能体实现

让我们通过一个完整的营销智能体案例来展示子图的实际应用：

4.4.1 内容生成子图

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
29
class ContentCreationSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def generate_content(state):
            prompt = f"""
            Target Audience: {state['audience']}
            Platform: {state['platform']}
            Campaign Goal: {state['goal']}
            """
            # 使用LLM生成内容
            content = generate_with_llm(prompt)
            state['generated_content'] = content
            return state
            
        def optimize_content(state):
            # 根据平台特点优化内容
            optimized = optimize_for_platform(
                state['generated_content'],
                state['platform']
            )
            state['final_content'] = optimized
            return state
            
        graph.add_node("generate", generate_content)
        graph.add_node("optimize", optimize_content)
        graph.add_edge("generate", "optimize")
        
        return graph

4.4.2 分析子图

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
class AnalyticsSubGraph(SubGraph):
    def build(self) -> Graph:
        graph = Graph()
        
        def analyze_performance(state):
            metrics = calculate_metrics(state['final_content'])
            state['analytics'] = {
                'engagement_score': metrics['engagement'],
                'reach_prediction': metrics['reach'],
                'conversion_estimate': metrics['conversion']
            }
            return state
            
        def generate_recommendations(state):
            recommendations = generate_improvements(
                state['analytics'],
                state['goal']
            )
            state['recommendations'] = recommendations
            return state
            
        graph.add_node("analyze", analyze_performance)
        graph.add_node("recommend", generate_recommendations)
        graph.add_edge("analyze", "recommend")
        
        return graph

4.4.3 主工作流程

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
def create_marketing_agent():
    main_graph = Graph()
    
    # 实例化子图
    content_graph = ContentCreationSubGraph()
    analytics_graph = AnalyticsSubGraph()
    
    # 添加配置节点
    def setup_campaign(state):
        # 初始化营销活动配置
        if 'config' not in state:
            state['config'] = {
                'audience': state.get('audience', 'general'),
                'platform': state.get('platform', 'twitter'),
                'goal': state.get('goal', 'engagement')
            }
        return state
    
    main_graph.add_node("setup", setup_campaign)
    main_graph.add_node("content", content_graph)
    main_graph.add_node("analytics", analytics_graph)
    
    # 构建工作流
    main_graph.add_edge("setup", "content")
    main_graph.add_edge("content", "analytics")
    
    return main_graph

4.5 最佳实践与注意事项

4.5.1 子图设计原则：

  • 保持子图功能单一
  • 确保清晰的输入输出接口
  • 适当处理状态传递

4.5.2 性能考虑：

  • 避免子图间频繁的大数据传输
  • 合理设计状态存储结构
  • 考虑异步处理需求

4.5.3 错误处理：

  • 在子图内部实现错误处理
  • 提供清晰的错误信息
  • 确保状态一致性

4.6 总结

LangGraph的子图架构为构建复杂AI工作流提供了强大而灵活的解决方案。通过合理使用子图，我们可以将复杂的任务分解为可管理的模块，提高代码的可维护性和可重用性。在实际应用中，子图不仅简化了开发过程，还为扩展和优化系统提供了便利。营销智能体的案例展示了如何在实践中运用这些概念，构建出功能强大、结构清晰的AI应用。

快来参与讨论吧~

博客近期分类标签归档
项目软件开发人工智能Staller
社交友链
更多关于本站
本站由 卡卡龙 使用 Stellar 1.29.1主题创建

本站访问量 次. 本文阅读量 次.

Stellar Copyright Licenses Hexo 备案 描述文字 描述文字