工具

from typing_extensions import Annotated
from langchain.agents.tool_node import InjectedState

# 访问当前对话状态
@tool
def summarize_conversation(
    state: Annotated[dict, InjectedState]
) -> str:
    """总结到目前为止的对话。"""
    messages = state["messages"]

    human_msgs = sum(1 for m in messages if m.__class__.__name__ == "HumanMessage")
    ai_msgs = sum(1 for m in messages if m.__class__.__name__ == "AIMessage")
    tool_msgs = sum(1 for m in messages if m.__class__.__name__ == "ToolMessage")

    return f"对话包含 {human_msgs} 条用户消息，{ai_msgs} 条 AI 响应，以及 {tool_msgs} 条工具结果"

# 访问自定义状态字段
@tool
def get_user_preference(
    pref_name: str,
    preferences: Annotated[dict, InjectedState("user_preferences")]  # InjectedState 参数对模型不可见
) -> str:
    """获取用户偏好值。"""
    return preferences.get(pref_name, "未设置")

from langgraph.types import Command
from langchain_core.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain_core.tools import tool, InjectedToolCallId
from typing_extensions import Annotated

# 通过删除所有消息来更新对话历史
@tool
def clear_conversation() -> Command:
    """清除对话历史。"""

    return Command(
        update={
            "messages": [RemoveMessage(id=REMOVE_ALL_MESSAGES)],
        }
    )

# 更新代理状态中的 user_name
@tool
def update_user_name(
    new_name: str,
    tool_call_id: Annotated[dict, InjectedToolCallId]
) -> Command:
    """更新用户名。"""
    return Command(update={"user_name": new_name})

from dataclasses import dataclass
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_core.tools import tool
from langgraph.runtime import get_runtime

USER_DATABASE = {
    "user123": {
        "name": "Alice Johnson",
        "account_type": "Premium",
        "balance": 5000,
        "email": "[email protected]"
    },
    "user456": {
        "name": "Bob Smith",
        "account_type": "Standard",
        "balance": 1200,
        "email": "[email protected]"
    }
}

@dataclass
class UserContext:
    user_id: str

@tool
def get_account_info() -> str:
    """获取当前用户的账户信息。"""
    runtime = get_runtime(UserContext)
    user_id = runtime.context.user_id

    if user_id in USER_DATABASE:
        user = USER_DATABASE[user_id]
        return f"账户持有人: {user['name']}\n类型: {user['account_type']}\n余额: ${user['balance']}"
    return "未找到用户"

model = ChatOpenAI(model="gpt-4o")
agent = create_agent(
    model,
    tools=[get_account_info],
    context_schema=UserContext,
    prompt="您是一位金融助手。"
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "我的当前余额是多少？"}]},
    context=UserContext(user_id="user123")
)

from langgraph.config import get_store

@tool
def get_user_info(user_id: str) -> str:
    """查找用户信息。"""
    store = get_store()
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "未知用户"

from typing import Any
from langgraph.config import get_store
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain_core.tools import tool

@tool
def get_user_info(user_id: str) -> str:
    """查找用户信息。"""
    store = get_store()
    user_info = store.get(("users",), user_id)
    return str(user_info.value) if user_info else "未知用户"

@tool
def save_user_info(user_id: str, user_info: dict[str, Any]) -> str:
    """保存用户信息。"""
    store = get_store()
    store.put(("users",), user_id, user_info)
    return "成功保存用户信息。"

store = InMemoryStore()
agent = create_agent(
    model,
    tools=[get_user_info, save_user_info],
    store=store
)

# 第一个会话：保存用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "保存以下用户：userid: abc123, name: Foo, age: 25, email: [email protected]"}]
})

# 第二个会话：获取用户信息
agent.invoke({
    "messages": [{"role": "user", "content": "获取 ID 为 'abc123' 的用户信息"}]
})
# 以下是 ID 为 "abc123" 的用户信息：
# - 姓名: Foo
# - 年龄: 25
# - 电子邮件: [email protected]