如何使用 LangGraph 实现生成式用户界面¶

import uuid
from typing import Annotated, Sequence, TypedDict

from langchain_core.messages import AIMessage, BaseMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.graph.ui import AnyUIMessage, ui_message_reducer, push_ui_message


class AgentState(TypedDict):  # noqa: D101
    messages: Annotated[Sequence[BaseMessage], add_messages]
    ui: Annotated[Sequence[AnyUIMessage], ui_message_reducer]


async def weather(state: AgentState):
    class WeatherOutput(TypedDict):
        city: str

    weather: WeatherOutput = (
        await ChatOpenAI(model="gpt-4o-mini")
        .with_structured_output(WeatherOutput)
        .with_config({"tags": ["nostream"]})
        .ainvoke(state["messages"])
    )

    message = AIMessage(
        id=str(uuid.uuid4()),
        content=f"Here's the weather for {weather['city']}",
    )

    # Emit UI elements associated with the message
    push_ui_message("weather", weather, message=message)
    return {"messages": [message]}


workflow = StateGraph(AgentState)
workflow.add_node(weather)
workflow.add_edge("__start__", "weather")
graph = workflow.compile()

from typing import Annotated, Sequence, TypedDict

from langchain_anthropic import ChatAnthropic
from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langgraph.graph.ui import AnyUIMessage, push_ui_message, ui_message_reducer


class AgentState(TypedDict):  # noqa: D101
    messages: Annotated[Sequence[BaseMessage], add_messages]
    ui: Annotated[Sequence[AnyUIMessage], ui_message_reducer]


class CreateTextDocument(TypedDict):
    """Prepare a document heading for the user."""

    title: str


async def writer_node(state: AgentState):
    model = ChatAnthropic(model="claude-3-5-sonnet-latest")
    message: AIMessage = await model.bind_tools(
        tools=[CreateTextDocument],
        tool_choice={"type": "tool", "name": "CreateTextDocument"},
    ).ainvoke(state["messages"])

    tool_call = next(
        (x["args"] for x in message.tool_calls if x["name"] == "CreateTextDocument"),
        None,
    )

    if tool_call:
        ui_message = push_ui_message("writer", tool_call, message=message)
        ui_message_id = ui_message["id"]

        # We're already streaming the LLM response to the client through UI messages
        # so we don't need to stream it again to the `messages` stream mode.
        content_stream = model.with_config({"tags": ["nostream"]}).astream(
            f"Create a document with the title: {tool_call['title']}"
        )

        content: AIMessageChunk | None = None
        async for chunk in content_stream:
            content = content + chunk if content else chunk

            push_ui_message(
                "writer",
                {"content": content.text()},
                id=ui_message_id,
                message=message,
                # Use `merge=rue` to merge props with the existing UI message
                merge=True,
            )

    return {"messages": [message]}

from langgraph.graph.ui import push_ui_message, delete_ui_message

# push message
message = push_ui_message("weather", {"city": "London"})

# remove said message
delete_ui_message(message["id"])