简介

在人工智能领域，检索增强生成（RAG）系统已成为处理简单查询和生成上下文相关响应的常用工具。然而，随着对更复杂的人工智能应用需求的增长，人们需要超越这些检索能力的系统。这就是人工智能代理--能够执行复杂的多步骤任务、在交互过程中保持状态并动态适应新信息的自主实体。LangGraph 是 LangChain 库的一个功能强大的扩展，旨在通过支持具有循环计算能力的有状态多角色应用程序，帮助开发人员构建这些高级人工智能代理。

在本文中，我们将探讨 LangGraph 如何改变人工智能开发，并通过一个计算太阳能电池板节能效果的示例，逐步介绍如何构建自己的人工智能代理。该示例将展示 LangGraph 的独特功能如何创建智能、适应性强、适用于现实世界的人工智能系统。

什么是 LangGraph？

LangGraph 是建立在 LangChain 基础上的高级库，旨在通过引入循环计算能力来增强大型语言模型（LLM）应用。LangChain 允许为线性工作流创建有向无环图（DAG），而 LangGraph 则在此基础上更进一步，允许添加循环，这对于开发复杂的、类似代理的行为至关重要。这些行为允许 LLM 在流程中不断循环，根据不断变化的条件动态决定下一步要采取的行动。

LangGraph 的核心是有状态图的概念：

• 状态： 代表在计算过程中维护和更新的上下文或内存。它确保图中的每一步都能访问前一步的相关信息，从而在整个过程中根据积累的数据做出动态决策。
• 节点： 作为图的构建模块，代表各个计算步骤或功能。每个节点都执行特定任务，如处理输入、做出决策或与外部系统交互。节点可以定制，以便在工作流程中执行各种操作。
• 边： 连接图中的节点，定义从一个步骤到下一个步骤的计算流程。它们支持条件逻辑，允许根据当前状态改变执行路径，并促进节点之间的数据和控制移动，从而实现复杂的多步骤工作流。

通过无缝管理图结构、状态和协调，LangGraph 重新定义了人工智能开发，使复杂的多角色应用程序的创建成为可能。通过自动状态管理，LangGraph 可确保在交互过程中保留上下文，使人工智能能够对不断变化的输入做出智能响应。它简化的代理协调可确保精确的执行和高效的信息交换，让你专注于设计创新的工作流程，而不是复杂的技术问题。LangGraph 的灵活性允许开发量身定制的高性能应用程序，其可扩展性和容错性确保你的系统即使在企业级也能保持稳健可靠。

分步指南

既然我们已经对 LangGraph 及其如何增强人工智能开发有了扎实的了解，那就让我们深入一个实际案例。在这个场景中，我们将建立一个人工智能代理，旨在根据用户输入计算太阳能电池板的潜在节能效果。这个代理可以作为太阳能电池板销售商网站上的潜在客户生成工具，与潜在客户互动，提供个性化的节能估算。通过收集每月电费等关键数据，该人工智能代理可以帮助客户了解太阳能的经济效益，同时为销售团队的后续工作提供潜在客户。该示例展示了 LangGraph 在创建智能动态系统方面的强大功能，该系统可自动执行复杂任务并推动业务价值。

第一步：导入必要的库

我们首先导入项目所需的所有基本 Python 库和模块。

from langchain_core.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import Runnable
from langchain_aws import ChatBedrock
import boto3
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import AnyMessage, add_messages
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableLambda
from langgraph.prebuilt import ToolNode

这些导入为利用 LangChain、LangGraph 和 AWS 服务构建我们的人工智能助手奠定了基础。

第二步：定义计算太阳能储蓄的工具

接下来，我们将定义一个工具，根据用户提供的每月电费来计算节能效果。

@tool
def compute_savings(monthly_cost: float) -> float:
    """
    Tool to compute the potential savings when switching to solar energy based on the user's monthly electricity cost.
    
    Args:
        monthly_cost (float): The user's current monthly electricity cost.
    
    Returns:
        dict: A dictionary containing:
            - 'number_of_panels': The estimated number of solar panels required.
            - 'installation_cost': The estimated installation cost.
            - 'net_savings_10_years': The net savings over 10 years after installation costs.
    """
    def calculate_solar_savings(monthly_cost):
        # Assumptions for the calculation
        cost_per_kWh = 0.28  
        cost_per_watt = 1.50  
        sunlight_hours_per_day = 3.5  
        panel_wattage = 350  
        system_lifetime_years = 10  
        # Monthly electricity consumption in kWh
        monthly_consumption_kWh = monthly_cost / cost_per_kWh
        
        # Required system size in kW
        daily_energy_production = monthly_consumption_kWh / 30
        system_size_kW = daily_energy_production / sunlight_hours_per_day
        
        # Number of panels and installation cost
        number_of_panels = system_size_kW * 1000 / panel_wattage
        installation_cost = system_size_kW * 1000 * cost_per_watt
        
        # Annual and net savings
        annual_savings = monthly_cost * 12
        total_savings_10_years = annual_savings * system_lifetime_years
        net_savings = total_savings_10_years - installation_cost
        
        return {
            "number_of_panels": round(number_of_panels),
            "installation_cost": round(installation_cost, 2),
            "net_savings_10_years": round(net_savings, 2)
        }
    # Return calculated solar savings
    return calculate_solar_savings(monthly_cost)

该函数处理用户的每月电费，并返回对太阳能电池板系统效益的详细估算，包括所需的电池板数量、安装成本和十年的净节省额。为简单起见，我们在计算中做了一些假设，如每千瓦时的平均成本和平均日照时数。不过，在更先进的人工智能代理版本中，我们可以直接从用户那里收集这些信息，从而更精确地根据用户的独特情况进行估算。

第三步：设置状态管理和错误处理

有效的状态管理和错误处理对于构建稳健的人工智能系统至关重要。在此，我们定义了管理错误和维护对话状态的实用程序。

def handle_tool_error(state) -> dict:
    """
    Function to handle errors that occur during tool execution.
    
    Args:
        state (dict): The current state of the AI agent, which includes messages and tool call details.
    
    Returns:
        dict: A dictionary containing error messages for each tool that encountered an issue.
    """
    # Retrieve the error from the current state
    error = state.get("error")
    
    # Access the tool calls from the last message in the state's message history
    tool_calls = state["messages"][-1].tool_calls
    
    # Return a list of ToolMessages with error details, linked to each tool call ID
    return {
        "messages": [
            ToolMessage(
                content=f"Error: {repr(error)}\n please fix your mistakes.",  # Format the error message for the user
                tool_call_id=tc["id"],  # Associate the error message with the corresponding tool call ID
            )
            for tc in tool_calls  # Iterate over each tool call to produce individual error messages
        ]
    }
def create_tool_node_with_fallback(tools: list) -> dict:
    """
    Function to create a tool node with fallback error handling.
    
    Args:
        tools (list): A list of tools to be included in the node.
    
    Returns:
        dict: A tool node that uses fallback behavior in case of errors.
    """
    # Create a ToolNode with the provided tools and attach a fallback mechanism
    # If an error occurs, it will invoke the handle_tool_error function to manage the error
    return ToolNode(tools).with_fallbacks(
        [RunnableLambda(handle_tool_error)],  # Use a lambda function to wrap the error handler
        exception_key="error"  # Specify that this fallback is for handling errors
    )

这些功能可确保在工具执行过程中遇到的任何错误都能得到妥善处理，并向用户提供有用的反馈。

第四步：定义状态和助手类

在这一步中，我们将定义人工智能代理如何管理其状态（正在进行的对话上下文），并确保它对用户的输入和工具的输出做出适当的响应。

为此，我们将使用 Python 的 TypedDict 创建一个状态类，以定义将要传递的消息的结构。状态将保存消息，包括来自用户的输入和来自代理或工具的输出。

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

接下来，我们创建助理类，它负责运行人工智能代理、与工具交互并管理对话流程。助理会调用工具，确保它们返回适当的结果，并处理执行过程中可能出现的任何重试或错误。其核心功能包括调用 Runnable（定义了调用 LLM 和工具（如 compute_savings）的流程），然后监控结果。如果代理没有返回有效的响应，或者工具没有提供有意义的数据，助理就会重新提示用户或要求澄清。它将继续在 Runnable 中循环，直到获得有效输出，从而确保顺利执行和有效响应。

class Assistant:
    def __init__(self, runnable: Runnable):
        # Initialize with the runnable that defines the process for interacting with the tools
        self.runnable = runnable
    def __call__(self, state: State):
        while True:
            # Invoke the runnable with the current state (messages and context)
            result = self.runnable.invoke(state)
            
            # If the tool fails to return valid output, re-prompt the user to clarify or retry
            if not result.tool_calls and (
                not result.content
                or isinstance(result.content, list)
                and not result.content[0].get("text")
            ):
                # Add a message to request a valid response
                messages = state["messages"] + [("user", "Respond with a real output.")]
                state = {**state, "messages": messages}
            else:
                # Break the loop when valid output is obtained
                break
        # Return the final state after processing the runnable
        return {"messages": result}

这种设置对于保持对话流畅并确保助手根据上下文做出适当回应至关重要。

第五步：使用 AWS Bedrock 设置 LLM

在这一步中，我们将使用 AWS Bedrock 配置大型语言模型 (LLM)，它将为人工智能助手的语言功能提供动力。AWS Bedrock 允许我们访问高级 LLM，如 Anthropic 的 Claude。要与 AWS 服务交互，你需要配置 AWS 凭据。这意味着你必须在环境中设置 AWS 凭据（通过 AWS CLI 或环境变量），或者使用 AWS SDK 可以访问的凭据文件。如果没有正确的 AWS 配置，助手将无法连接到 Bedrock 等 AWS 服务来运行 LLM。

def get_bedrock_client(region):
    return boto3.client("bedrock-runtime", region_name=region)
def create_bedrock_llm(client):
    return ChatBedrock(model_id='anthropic.claude-3-sonnet-20240229-v1:0', client=client, model_kwargs={'temperature': 0}, region_name='us-east-1')
llm = create_bedrock_llm(get_bedrock_client(region='us-east-1'))

这种集成可确保助手能有效解释和响应用户输入。

第六步：定义助手的工作流程

现在我们已经建立了 LLM 和工具，下一步就是定义人工智能助手的工作流程。这包括创建对话模板、指定助手将使用的工具，以及配置人工智能代理如何响应用户输入和触发不同功能（如计算太阳能储蓄）。工作流程本质上是管理助手如何与用户互动、收集信息以及调用工具提供结果的逻辑。

工作流程的第一部分包括创建一个提示模板，定义助手将如何与用户交流。提示有助于指导人工智能助手确定向用户询问什么、如何根据输入做出回应以及何时触发计算节省等工具。

在本例中，助手需要询问用户每月的电费，以计算太阳能电池板的节省额。我们是这样定义对话的：

primary_assistant_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system","system",
            '''You are a helpful customer support assistant for Solar Panels Belgium.
            You should get the following information from them:
            - monthly electricity cost
            If you are not able to discern this info, ask them to clarify! Do not attempt to wildly guess.
            After you are able to discern all the information, call the relevant tool.
            ''',
        ),
        ("placeholder", "{messages}"),
    ]
)

• 系统信息： 这条信息是人工智能代理的指南，指示它询问用户每月的电费，如果信息不清楚，就不要猜测。助手会不断提示用户，直到收集到所需数据。
• 占位符： 这个占位符允许助手动态地注入对话中的信息，创建一个持续的对话，用户的输入会影响下一步的操作。

接下来，我们定义了助手在交互过程中将使用的工具，其中最主要的工具是计算节省额度，它可以根据用户每月的电费计算出潜在的节省额度。在列表中指定工具后，我们使用 llm.bind_tools() 方法将它们绑定到助手的工作流中。这一步骤可确保人工智能助手在对话过程中根据需要访问和触发工具，从而在用户和助手之间创建无缝交互。

# Define the tools the assistant will use
part_1_tools = [
    compute_savings
]
# Bind the tools to the assistant's workflow
part_1_assistant_runnable = primary_assistant_prompt | llm.bind_tools(part_1_tools)

第七步：构建图结构

在这一步中，我们使用 LangGraph 构建人工智能助手的图结构，它控制着助手处理用户输入、触发工具以及在不同阶段之间移动的方式。图为核心操作（如调用助手和工具）定义了节点，并为这些节点之间的流动定义了边。

LangGraph 中的每个节点代表一个操作步骤，如与用户交互或执行工具。我们为这个人工智能助手定义了两个关键节点：

• 助理节点： 管理对话流程，询问用户电费并处理回复。
• 工具节点： 执行工具（例如：compute_savings），计算用户太阳能电池板节省的电费。

builder = StateGraph(State)
builder.add_node("assistant", Assistant(part_1_assistant_runnable))"assistant", Assistant(part_1_assistant_runnable))
builder.add_node("tools", create_tool_node_with_fallback(part_1_tools))

边定义了节点之间的流动方式。在这里，助手开始对话，然后在收集到所需输入后转到工具，并在工具执行后返回助手。

builder.add_edge(START, "assistant")  # Start with the assistant"assistant")  # Start with the assistant
builder.add_conditional_edges("assistant", tools_condition)  # Move to tools after input
builder.add_edge("tools", "assistant")  # Return to assistant after tool execution

我们使用记忆保护器（MemorySaver）来确保图形在不同步骤中保留对话状态。这样，助手就能记住用户的输入，确保多步骤交互的连续性。

memory = MemorySaver()
graph = builder.compile(checkpointer=memory)compile(checkpointer=memory)

第八步：运行助手

最后，你可以通过启动图表和开始对话来运行助手。

# import shutil
import uuid
# Let's create an example conversation a user might have with the assistant
tutorial_questions = [
    'hey',
    'can you calculate my energy saving',
    "my montly cost is $100, what will i save"
]
# Update with the backup file so we can restart from the original place in each section
# shutil.copy(backup_file, db)
thread_id = str(uuid.uuid4())
config = {
    "configurable": {
        # The passenger_id is used in our flight tools to
        # fetch the user's flight information
        # "passenger_id": "3442 587242",
        # Checkpoints are accessed by thread_id
        "thread_id": thread_id,
    }
}
_printed = set()
for question in tutorial_questions:
    events = graph.stream(
        {"messages": ("user", question)}, config, stream_mode="values"
    )
    for event in events:
        _print_event(event, _printed)

结论

按照以上步骤，你已经使用 LangGraph 成功创建了一个人工智能助手，它可以根据用户输入计算太阳能电池板的节能效果。本教程展示了 LangGraph 在管理复杂、多步骤流程方面的强大功能，并重点介绍了如何利用先进的人工智能工具高效地解决现实世界中的难题。