Langgraph实战-Agent-使用ReAct进行情感分析

概述

在这个例子中创建了一个情感分析 Agent，它使用 LLM（大型语言模型）来处理客户反馈，并集成自定义情感分析工具，将反馈分为正面、中性或负面。根据情感分析结果， Agent使用 ReAct（推理、行动、观察）框架生成适当的回应。 Agent对客户反馈进行推理，调用情感分析工具，并作出回应。

关键点

1. 用于推理的 LLM（LLM for Reasoning）： LLM 处理客户反馈，并决定是调用情感分析工具还是直接做出回应。
2. 情感分析工具（Sentiment Analysis Tool）： 自定义情感分析逻辑（使用 TextBlob），以确定情感是积极、中性还是消极。
3. 状态图（StateGraph）： Agent的工作流程通过状态图进行管理，状态图整合了推理、行动（情感分析）和响应。

代码实现逻辑

1. 定义 Agent状态AgentState：该状态跟踪客户反馈和消息序列。
2. 定义情绪分析工具：我们使用 TextBlob 实现了一个自定义工具来分析客户反馈情绪。
3. 定义响应生成工具（respond_based_on_sentiment）：一旦分析了情绪，我们就会使用第二个工具根据情绪生成回应。
4. 将工具绑定到 LLM（bind_tools）：我们将情绪分析和响应工具绑定到 LLM，以便它可以在需要时调用这些工具。
5. 定义工具执行节点（tool_node）：我们定义一个节点，当确定应该调用某个工具时，该节点会执行该工具。
6. 定义推理节点（添加agent节点）：此节点使用 LLM 来推理客户反馈，并直接响应或调用工具。
7. 定义状态转换（should_continue）：我们定义了一个条件来确定是否应该继续调用工具或提供最终响应。
8. 构建工作流图，并添加处理的边。
9. 然后就可以开始使用该Agent了。

完整代码

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import json
from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END, START
from display_graph import display_graph
import os

class AgentState(TypedDict):
    """The state of the agent."""
    # `add_messages` is a reducer that manages the message sequence.
    messages: Annotated[Sequence[BaseMessage], add_messages]

@tool
def analyze_sentiment(feedback: str) -> str:
    """Analyze customer feedback sentiment with below custom logic"""
    from textblob import TextBlob
    analysis = TextBlob(feedback)
    if analysis.sentiment.polarity > 0.5:
        return "positive"
    elif analysis.sentiment.polarity == 0.5:
        return "neutral"
    else:
        return "negative"

@tool
def respond_based_on_sentiment(sentiment: str) -> str:
    """Only respond as below to the customer based on the analyzed sentiment."""
    if sentiment == "positive":
        return "Thank you for your positive feedback!"
    elif sentiment == "neutral":
        return "We appreciate your feedback."
    else:
        return "We're sorry to hear that you're not satisfied. How can we help?"

# 定义工具
tools = [analyze_sentiment, respond_based_on_sentiment]
# 获取工具名
tools_by_name = {tool.name: tool for tool in tools}

# llm = ChatOpenAI(model="gpt-4o-mini")
# # 通过 ChatOpenAI 初始化一个 LLM 模型
local_llm = "Qwen/Qwen2.5-7B-Instruct"
base_url = "https://api.siliconflow.cn/v1"
llm = ChatOpenAI(model=local_llm, base_url=base_url, api_key="sk-xx")

# 工具绑定
llm = llm.bind_tools(tools)
tools_by_name = {tool.name: tool for tool in tools}


def tool_node(state: AgentState):
    outputs = []
    for tool_call in state["messages"][-1].tool_calls:
        tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
        outputs.append(
            ToolMessage(
                content=json.dumps(tool_result),
                name=tool_call["name"],
                tool_call_id=tool_call["id"],
            )
        )
    return {"messages": outputs}

def call_model(state: AgentState, config: RunnableConfig):
    system_prompt = SystemMessage(
        content="You are a helpful assistant tasked with responding to customer feedback."
    )
    response = llm.invoke([system_prompt] + state["messages"], config)
    return {"messages": [response]}

def should_continue(state: AgentState):
    last_message = state["messages"][-1]
    # If there is no tool call, then we finish
    if not last_message.tool_calls:
        return "end"
    else:
        return "continue"

# 构建工作流图
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END,
    },
)
workflow.add_edge("tools", "agent")

graph = workflow.compile()


# 根据用户反馈初始化用户状态
initial_state = {"messages": [("user", "The product was great but the delivery was poor.")]}

# 打印会话过程
def print_stream(stream):
    for s in stream:
        message = s["messages"][-1]
        if isinstance(message, tuple):
            print(message)
        else:
            message.pretty_print()

# 运行agent
print_stream(graph.stream(initial_state, stream_mode="values"))

结果输出

================================ Human Message =================================

The product was great but the delivery was poor.
================================== Ai Message ==================================
Tool Calls:
  analyze_sentiment (01965ff7ea608dcf162f95cd988aa607)
 Call ID: 01965ff7ea608dcf162f95cd988aa607
  Args:
    feedback: The product was great but the delivery was poor.
================================= Tool Message =================================
Name: analyze_sentiment

"negative"
================================== Ai Message ==================================
Tool Calls:
  respond_based_on_sentiment (01965ff7f1f30ed571e49350de8b530b)
 Call ID: 01965ff7f1f30ed571e49350de8b530b
  Args:
    sentiment: negative
================================= Tool Message =================================
Name: respond_based_on_sentiment

"We're sorry to hear that you're not satisfied. How can we help?"
================================== Ai Message ==================================

We're sorry to hear that you're not satisfied. How can we help?

总结

可以根据该例子进行扩展，从而去适配不同的业务场景。这里要注意的是，需要先确定业务场景，然后根据业务逻辑来规划工作流图，然后确定各个工具的功能。