自定义 Tool：从函数封装到带状态写回的执行接口

from langchain.tools import tool


@tool
def get_weather(city: str) -> str:
    """查询指定城市天气。"""
    return f"{city} 当前天气晴。"

from typing import Literal
from pydantic import BaseModel, Field
from langchain.tools import tool


class SearchTicketsInput(BaseModel):
    query: str = Field(description="要搜索的问题描述")
    limit: int = Field(default=5, ge=1, le=20, description="返回数量")
    source: Literal["incident", "knowledge_base"] = Field(
        default="incident",
        description="查询来源"
    )


@tool(args_schema=SearchTicketsInput)
def search_tickets(query: str, limit: int = 5, source: str = "incident") -> str:
    """搜索工单或知识库中的相关记录。"""
    return f"从 {source} 中找到 {limit} 条与 {query} 相关的记录"

from dataclasses import dataclass
from langchain.tools import tool, ToolRuntime


@dataclass
class Context:
    user_id: str


@tool
def fetch_user_preferences(runtime: ToolRuntime[Context]) -> str:
    """读取当前用户偏好。"""
    user_id = runtime.context.user_id

    if runtime.store:
        memory = runtime.store.get(("users",), user_id)
        if memory:
            return memory.value["preferences"]

    return "用户未设置偏好"

from langgraph.types import Command
from langchain.tools import tool


@tool
def classify_priority(score: int) -> Command:
    """根据分数设置工单优先级。"""
    priority = "high" if score >= 80 else "normal"
    return Command(update={"ticket_priority": priority})

from langchain.tools import tool
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph import StateGraph, MessagesState, START


@tool
def search(query: str) -> str:
    """搜索资料。"""
    return f"Results for: {query}"


tool_node = ToolNode([search], handle_tool_errors=True)


def call_llm(state: MessagesState):
    ...


builder = StateGraph(MessagesState)
builder.add_node("llm", call_llm)
builder.add_node("tools", tool_node)
builder.add_edge(START, "llm")
builder.add_conditional_edges("llm", tools_condition)
builder.add_edge("tools", "llm")
graph = builder.compile()