多意图智能体 (Multi Intention Agent)

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode
from langchain_core.messages import SystemMessage, HumanMessage
from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool
from typing import Literal

_ = load_dotenv()

# 配置大模型服务
llm = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    model="qwen3-coder-plus",
    temperature=0.7,
)

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气信息"""
    return f"{city}的天气是晴天，温度25°C，适合出行！"

@tool
def calculate(expression: str) -> str:
    """计算数学表达式的结果"""
    try:
        result = eval(expression)
        return f"计算结果：{expression} = {result}"
    except Exception as e:
        return f"计算错误：{str(e)}"

# 创建工具节点封装
weather_tools = [get_weather]
math_tools = [calculate]

weather_tool_node = ToolNode(weather_tools)
math_tool_node = ToolNode(math_tools)

# === 1. 意图分类节点 ===
def classify_intention(state: MessagesState, config: RunnableConfig):
    """使用LLM对用户问题进行意图分类"""
    system_prompt = """你是一个意图分类助手。请根据用户的问题，判断其意图类型。

    可选的意图类型包括：
    1. weather - 天气相关的问题（如查询天气、温度等）
    2. math - 数学计算相关的问题（如计算、数学运算等）
    3. chat - 通用聊天对话（其他所有问题）

    请只返回一个单词：weather、math 或 chat，不要返回其他内容。"""

    last_message = state["messages"][-1]
    classification_prompt = f"""用户问题：{last_message.content}

    请判断意图类型（只返回一个单词：weather、math 或 chat）："""

    messages = [
        SystemMessage(content=system_prompt),
        HumanMessage(content=classification_prompt)
    ]
    
    response = llm.invoke(messages)
    intention = response.content.strip().lower()
    
    if intention not in ["weather", "math", "chat"]:
        intention = "chat"
    
    # 将分类结果作为一条特殊消息存入历史
    classification_msg = HumanMessage(
        content=f"[系统分类结果：{intention}]"
    )
    
    return {"messages": state["messages"] + [classification_msg]}

# === 2. 各个 Handler 节点 ===

def weather_handler(state: MessagesState, config: RunnableConfig):
    """处理天气相关的问题"""
    system_prompt = "你是一个天气助手，可以帮助用户查询天气信息。"
    all_messages = [SystemMessage(content=system_prompt)] + state["messages"]
    model = llm.bind_tools(weather_tools)
    return {"messages": [model.invoke(all_messages)]}

def math_handler(state: MessagesState, config: RunnableConfig):
    """处理数学计算相关的问题"""
    system_prompt = "你是一个数学计算助手，可以帮助用户进行数学计算。"
    all_messages = [SystemMessage(content=system_prompt)] + state["messages"]
    model = llm.bind_tools(math_tools)
    return {"messages": [model.invoke(all_messages)]}

def chat_handler(state: MessagesState, config: RunnableConfig):
    """处理通用聊天对话"""
    system_prompt = "你是一个友好的助手，可以回答各种问题并进行对话。"
    all_messages = [SystemMessage(content=system_prompt)] + state["messages"]
    return {"messages": [llm.invoke(all_messages)]}

def route_by_intention(state: MessagesState, config: RunnableConfig) -> Literal["weather", "math", "chat"]:
    """根据分类结果路由到不同的处理节点"""
    last_message = state["messages"][-1].content
    
    if "[系统分类结果：weather]" in last_message:
        return "weather"
    elif "[系统分类结果：math]" in last_message:
        return "math"
    else:
        return "chat"

def should_use_tool(state: MessagesState, config: RunnableConfig):
    """通用判断：是否需要调用工具"""
    last_message = state["messages"][-1]
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls:
        return "use_tool"
    return "end"

def build_multi_intention_graph():
    builder = StateGraph(MessagesState)
    
    # 添加节点
    builder.add_node("classify", classify_intention)
    builder.add_node("weather_handler", weather_handler)
    builder.add_node("math_handler", math_handler)
    builder.add_node("chat_handler", chat_handler)
    builder.add_node("weather_tool", weather_tool_node)
    builder.add_node("math_tool", math_tool_node)
    
    # 流程开始 -> 分类
    builder.add_edge(START, "classify")
    
    # 分类 -> 路由
    builder.add_conditional_edges(
        "classify",
        route_by_intention,
        {
            "weather": "weather_handler",
            "math": "math_handler",
            "chat": "chat_handler",
        },
    )
    
    # 天气处理逻辑
    builder.add_conditional_edges(
        "weather_handler",
        should_use_tool,
        {"use_tool": "weather_tool", "end": END},
    )
    builder.add_edge("weather_tool", "weather_handler")
    
    # 数学处理逻辑
    builder.add_conditional_edges(
        "math_handler",
        should_use_tool,
        {"use_tool": "math_tool", "end": END},
    )
    builder.add_edge("math_tool", "math_handler")
    
    # 聊天直接结束
    builder.add_edge("chat_handler", END)
    
    return builder.compile()

# 初始化图
graph = build_multi_intention_graph()

# 可视化（可选，需要 graphviz）
try:
    print(graph.get_graph().draw_ascii())
    # 确保 img 目录存在
    pathlib.Path("img").mkdir(exist_ok=True)
    png_bytes = graph.get_graph().draw_mermaid_png()
    with open("./img/graph_viz.png", "wb") as f:
        f.write(png_bytes)
    print("图结构图片已保存至 ./img/graph_viz.png")
except Exception as e:
    print(f"可视化生成失败 (不影响运行): {e}")

def run_demo(query: str):
    print(f"\n\n🚀 用户问题：{query}")
    print("=" * 40)
    
    response = graph.invoke({"messages": [HumanMessage(content=query)]})
    
    for msg in response["messages"]:
        # 简单过滤一下，只打印内容
        content = msg.content
        if content:
            role = msg.__class__.__name__
            print(f"[{role}]: {content}")
    print("=" * 40)

# 测试用例
run_demo("帮我计算一下 123 + 456 等于多少？")
run_demo("北京今天天气怎么样？")
run_demo("你好，请介绍一下你自己。")