LangGraph 单位换算智能体示例

作者：追风剑情发布于：2026-5-7 13:05 分类：AI
安装依赖
pip install -U langgraph langchain
import os
os.environ["LANGGRAPH_ALLOWED_OBJECTS"] = "core"

import warnings
warnings.filterwarnings("ignore", message="The default value of `allowed_objects` will change")

import re
from typing import TypedDict, Optional
from langgraph.graph import StateGraph, END
from langchain.tools import tool

# ========== 状态定义 ==========
class GraphState(TypedDict):
    input_query: str
    raw_result: str
    formatted_result: str

# ========== 单位换算数据 ==========
UNITS = {
    "米": 1, "公尺": 1, "m": 1, "meter": 1, "meters": 1,
    "英尺": 0.3048, "ft": 0.3048, "foot": 0.3048, "feet": 0.3048, "呎": 0.3048,
    "公里": 1000, "km": 1000, "kilometer": 1000,
    "英里": 1609.34, "mile": 1609.34, "mi": 1609.34,
    "厘米": 0.01, "cm": 0.01, "公分": 0.01,
    "毫米": 0.001, "mm": 0.001,
    "码": 0.9144, "yard": 0.9144, "yd": 0.9144,
}

FROM_METER_FACTOR = {
    "米": 1, "公尺": 1, "m": 1, "meter": 1, "meters": 1,
    "英尺": 3.28084, "ft": 3.28084, "foot": 3.28084, "feet": 3.28084, "呎": 3.28084,
    "公里": 0.001, "km": 0.001,
    "英里": 0.000621371, "mile": 0.000621371, "mi": 0.000621371,
    "厘米": 100, "cm": 100, "公分": 100,
    "毫米": 1000, "mm": 1000,
    "码": 1.09361, "yard": 1.09361, "yd": 1.09361,
}

# ========== 解析函数 ==========
def parse_conversion(query: str) -> Optional[tuple[float, str, str]]:
    """解析用户查询，返回 (数值, 源单位, 目标单位)"""
    cleaned = re.sub(r'[？?！!。，,、\s]', '', query)
    num_match = re.search(r'(\d+(?:\.\d+)?)', cleaned)
    if not num_match:
        return None
    value = float(num_match.group(1))
    rest = cleaned[num_match.end():]
    
    matches = []
    for unit in sorted(UNITS.keys(), key=len, reverse=True):
        for m in re.finditer(re.escape(unit), rest):
            start, end = m.start(), m.end()
            matches.append((start, end, unit))
    matches.sort(key=lambda x: x[1]-x[0], reverse=True)
    selected = []
    occupied = set()
    for start, end, unit in matches:
        if any(pos in occupied for pos in range(start, end)):
            continue
        selected.append((start, unit))
        occupied.update(range(start, end))
    selected.sort(key=lambda x: x[0])
    if len(selected) < 2:
        return None
    
    keywords = ["等于多少", "等于", "转换为", "转成", "换算成", "转换成", "到", "是多少", "为"]
    kw_pos = len(rest)
    for kw in keywords:
        pos = rest.find(kw)
        if pos != -1:
            kw_pos = pos
            break
    
    src_unit = tgt_unit = None
    for pos, unit in selected:
        if pos < kw_pos:
            src_unit = unit
        else:
            tgt_unit = unit
    if src_unit is None and tgt_unit is None and len(selected) >= 2:
        src_unit, tgt_unit = selected[0][1], selected[1][1]
    
    if src_unit is None or tgt_unit is None:
        return None
    return value, src_unit, tgt_unit

# ========== 工具函数 ==========
@tool(description="长度单位换算，例如「15米等于多少英尺」")
def unit_converter(query: str) -> str:
    """执行单位换算"""
    parsed = parse_conversion(query)
    if parsed is None:
        return "无法解析单位换算请求，请使用如：'15米等于多少英尺' 或 '10公里转换为英里'"
    
    value, src_unit, tgt_unit = parsed
    if src_unit not in UNITS:
        return f"不支持的单位：{src_unit}"
    if tgt_unit not in FROM_METER_FACTOR:
        return f"不支持的单位：{tgt_unit}"
    
    value_in_meters = value * UNITS[src_unit]
    result = value_in_meters * FROM_METER_FACTOR[tgt_unit]
    return f"{value:.2f} {src_unit} = {result:.2f} {tgt_unit}"

# ========== LangGraph 节点 ==========
def process_query(state: GraphState) -> GraphState:
    raw = unit_converter.invoke({"query": state["input_query"]})
    state["raw_result"] = raw
    return state

def format_result(state: GraphState) -> GraphState:
    raw = state["raw_result"]
    if "=" in raw:
        a, b = raw.split("=")
        state["formatted_result"] = f"换算结果：{a.strip()} = {b.strip()}"
    else:
        state["formatted_result"] = f"信息：{raw}"
    return state

def build_conversion_graph():
    workflow = StateGraph(GraphState)
    workflow.add_node("process", process_query)
    workflow.add_node("format", format_result)
    workflow.set_entry_point("process")
    workflow.add_edge("process", "format")
    workflow.add_edge("format", END)
    return workflow.compile()

# ========== 测试入口 ==========
if __name__ == "__main__":
    app = build_conversion_graph()
    
    test_queries = [
        "15米等于多少英尺",
        "10公里转换为英里",
        "100厘米等于多少米",
        "3英尺到米",
        "5000毫米转换成米",
    ]
    
    print("=" * 50)
    print("单位换算智能体测试")
    print("=" * 50)
    
    for q in test_queries:
        print(f"\n输入：{q}")
        initial = {"input_query": q, "raw_result": "", "formatted_result": ""}
        result = app.invoke(initial)
        print(result["formatted_result"])
    
    print("\n" + "=" * 50)
    print("交互模式（输入 '退出' 结束）")
    while True:
        user = input("\n请输入换算需求：")
        if user in ["退出", "quit", "exit"]:
            print("已退出")
            break
        initial = {"input_query": user, "raw_result": "", "formatted_result": ""}
        res = app.invoke(initial)
        print(res["formatted_result"])
运行测试