使用LangGraph开发AI智能体。

AI领域正从基础的RAG系统向更智能的AI智能体进化，后者能处理更复杂的任务并适应新信息。LangGraph作为LangChain库的扩展，助力开发者构建具有状态管理和循环计算能力的先进AI系统。本文教大家如何使用LangGraph开发一个太阳能节能计算的智能体。

1 LangGraph概述

LangGraph是LangChain的高级库，为大型语言模型（LLM）带来循环计算能力。它超越了LangChain的线性工作流，通过循环支持复杂的任务处理。

• 状态：维护计算过程中的上下文，实现基于累积数据的动态决策。
• 节点：代表计算步骤，执行特定任务，可定制以适应不同工作流。
• 边：连接节点，定义计算流程，支持条件逻辑，实现复杂工作流。
  
  

LangGraph简化了AI开发，自动管理状态，保持上下文，使AI能智能响应变化。它让开发者专注于创新，而非技术细节，同时确保应用程序的高性能和可靠性。

2 逐步指南

了解LangGraph后，我们通过实例来实际应用：构建一个AI智能体，用于计算太阳能板节能潜力，并在销售网站上与潜在客户互动，提供个性化节能估算。这个智能体有助于教育客户太阳能的经济效益，并筛选出值得跟进的潜在客户。

步骤1：导入必要的库

我们先导入构建AI助手所需的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
from langgraph.prebuilt import tools_condition

这些库的导入为我们利用LangChain、LangGraph和AWS服务来构建AI助手打下了坚实的基础。

步骤2：定义计算太阳能节省的工具

接下来，定义一个工具来根据用户每月的电费计算使用太阳能板可能节省的费用。

@tool
def compute_savings(monthly_cost: float) -> float:
    """
    根据用户每月电费计算切换到太阳能时潜在节省的工具。
    
    参数：
        monthly_cost (float)：用户当前的每月电费。
    
    返回：
        dict：包含以下内容的字典：
            - 'number_of_panels'：估计所需的太阳能板数量。
            - 'installation_cost'：估计的安装成本。
            - 'net_savings_10_years'：安装成本后的10年净节省。
    """
    def calculate_solar_savings(monthly_cost):
        # 计算的假设
        cost_per_kWh = 0.28  
        cost_per_watt = 1.50  
        sunlight_hours_per_day = 3.5  
        panel_wattage = 350  
        system_lifetime_years = 10  

        # 每月用电量（千瓦时）
        monthly_consumption_kWh = monthly_cost / cost_per_kWh
        
        # 所需系统大小（千瓦）
        daily_energy_production = monthly_consumption_kWh / 30
        system_size_kW = daily_energy_production / sunlight_hours_per_day
        
        # 太阳能板数量和安装成本
        number_of_panels = system_size_kW * 1000 / panel_wattage
        installation_cost = system_size_kW * 1000 * cost_per_watt
        
        # 年度和净节省
        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 calculate_solar_savings(monthly_cost)

这个函数基于用户电费数据，提供太阳能板系统的详细节省估算，包括所需板数量、安装成本和未来十年的净节省。目前，我们使用了一些平均值来进行简化计算。未来，我们可以从用户那里直接获取更精确的数据，以提供更个性化的估算。

步骤3：设置状态管理和错误处理

有效的状态管理和错误处理对于构建健壮的AI系统非常重要。在这里，我们定义了工具来管理错误并维护对话的状态。

def handle_tool_error(state) -> dict:
    """
    处理工具执行期间发生的错误的函数。
    
    参数：        state (dict)：AI智能体的当前状态，包括消息和工具调用详情。
    
    返回：
        dict：包含每个遇到问题的工具的错误消息的字典。
    """
    # 从当前状态中检索错误
    error = state.get("error")
    
    # 从状态的消息历史中获取最后一个消息的工具调用
    tool_calls = state["messages"][-1].tool_calls
    
    # 返回包含错误详情的ToolMessages列表，与每个工具调用ID关联
    return {
        "messages": [
            ToolMessage(
                content=f"错误：{repr(error)}\n请修正你的错误。",  # 为用户格式化错误消息
                tool_call_id=tc["id"],  # 将错误消息与相应的工具调用ID关联
            )
            for tc in tool_calls  # 遍历每个工具调用以产生单独的错误消息
        ]
    }

def create_tool_node_with_fallback(tools: list) -> dict:
    """
    创建具有后备错误处理的工具节点的函数。
    
    参数：
        tools (list)：要包含在节点中的工具列表。
    
    返回：
        dict：如果发生错误，使用后备行为的工具节点。
    """
    # 使用提供的工具创建ToolNode，并附加后备机制
    # 如果发生错误，将调用handle_tool_error函数来管理错误
    return ToolNode(tools).with_fallbacks(
        [RunnableLambda(handle_tool_error)],  # 使用lambda函数包装错误处理器
        exception_key="error"  # 指定这个后备是用于处理错误的
    )

这些函数确保在工具执行期间遇到的任何错误都能得到优雅地处理，为用户提供有用的反馈。

步骤4：定义状态和助手类

在此步骤，我们设定AI智能体如何维护对话状态并响应用户输入及工具输出。

用Python的TypedDict创建State类来规范消息结构，包括用户和系统的消息。

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

然后，构建助手类来驱动AI智能体，管理对话。助手类调用工具，处理结果，并在需要时重新向用户询问。它通过循环调用Runnable直至获得有效输出，确保对话流畅。

class Assistant:
    def __init__(self, runnable: Runnable):
        # 使用定义与工具交互过程的可运行对象进行初始化
        self.runnable = runnable

    def
 __call__(self, state: State):
        while True:
            # 使用当前状态（消息和上下文）调用可运行对象
            result = self.runnable.invoke(state)
            
            # 如果工具未能返回有效输出，重新提示用户澄清或重试
            if not result.tool_calls and (
                not result.content
                or isinstance(result.content, list)
                and not result.content[0].get("text")
            ):
                # 添加请求有效响应的消息
                messages = state["messages"] + [("user", "请给出一个真实的输出。")]
                state = {**state, "messages": messages}
            else:
                # 当获得有效输出时跳出循环
                break

        # 在处理完可运行对象后返回最终状态
        return {"messages": result}

这个机制确保了对话的连贯性和助手的恰当响应。

步骤5：使用AWS Bedrock配置LLM

在这一步，我们通过AWS Bedrock设置大型语言模型（LLM），增强AI助手的语言处理能力。需要先配置AWS凭证，以便访问Bedrock服务。

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'))

这确保了助手能准确理解和回应用户。

步骤6：定义助手的工作流程

设置好LLM和工具后，现在定义AI助手的工作流程，主要涉及创建对话模板和指定工具的使用。

工作流程的第一部分创建一个模板来引导助手与用户沟通，明确要问的问题和如何根据回答调用工具。

primary_assistant_prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            '''你是太阳能板比利时的乐于助人的客服助手。
            你应该从他们那里获取以下信息：
            - 每月电费
            如果你不能清晰地识别这些信息，要求他们澄清！不要试图胡乱猜测。

            在你能够清晰地识别所有信息后，调用相关工具。
            ''',
        ),
        ("placeholder", "{messages}"),
    ]
)

确定助手将使用的工具，如compute_savings，并将其绑定到工作流程中，以便在对话中适时调用。

# 定义助手将使用的工具
part_1_tools = [
    compute_savings
]

# 将工具绑定到助手的工作流程
part_1_assistant_runnable = primary_assistant_prompt | llm.bind_tools(part_1_tools)

这确保了助手能根据用户输入灵活响应，实现顺畅的对话体验。

步骤7：构建图结构

利用LangGraph，我们为AI助手搭建图结构，控制其处理用户输入、触发工具和阶段转换的流程。

• 节点：代表操作步骤，如助手互动和工具执行。
• 边：定义步骤间的流程，如从助手到工具，再返回助手。

AI智能体旨在计算太阳能板潜在节能节省。

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

边定义了流程如何在节点之间移动。在这里，助手开始对话，一旦收集到所需输入，就过渡到工具，并在工具执行后返回助手。

builder.add_edge(START, "assistant")  # 从助手开始
builder.add_conditional_edges("assistant", tools_condition)  # 输入后移动到工具
builder.add_edge("tools", "assistant")  # 工具执行后返回助手

我们使用MemorySaver确保图在不同步骤之间保留对话状态。这允许助手记住用户的输入，确保多步骤交互中的连续性。

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

步骤8：运行助手

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

# import shutil
import uuid

# 让我们创建一个用户可能与助手进行的示例对话
tutorial_questions = [
    '嘿',
    '你能计算我的节能吗',
    “我的每月成本是100美元，我能节省多少”
]

thread_id = str(uuid.uuid4())

config = {
    "configurable": {
        "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成功打造了一个能根据用户输入计算太阳能节能的AI助手。这凸显了LangGraph在处理复杂任务和解决实际问题中的优势。