如何异步运行图¶

%%capture --no-stderr
%pip install --quiet -U langgraph langchain_anthropic

import getpass
import os


def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph.message import add_messages

# 添加消息本质上是用更多内容来实现这一点。
# 稳健处理
# def add_messages(left: list, right: list):
#     返回左边 + 右边


class State(TypedDict):
    messages: Annotated[list, add_messages]

from langchain_core.tools import tool


@tool
def search(query: str):
    """呼叫浏览网页。"""
    # 这只是一个占位符，但不要告诉语言模型这个…
    return ["The answer to your question lies within."]


tools = [search]

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools)

from langchain_anthropic import ChatAnthropic

model = ChatAnthropic(model="claude-3-haiku-20240307")

model = model.bind_tools(tools)

from typing import Literal


# 定义一个决定是否继续的函数。
def should_continue(state: State) -> Literal["end", "continue"]:
    messages = state["messages"]
    last_message = messages[-1]
    # 如果没有工具调用，那么我们就结束。
    if not last_message.tool_calls:
        return "end"
    # 否则如果有，我们继续。
    else:
        return "continue"


# 定义调用模型的函数。
async def call_model(state: State):
    messages = state["messages"]
    response = await model.ainvoke(messages)
    # 我们返回一个列表，因为这将被添加到现有列表中。
    return {"messages": [response]}

from langgraph.graph import END, StateGraph, START

# 定义一个新图。
workflow = StateGraph(State)

# 定义我们将循环的两个节点。
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)

# 将入口点设置为 `agent`。
# 这意味着这个节点是第一个被调用的。
workflow.add_edge(START, "agent")

# 我们现在添加一个条件边。
workflow.add_conditional_edges(
    # 首先，我们定义起始节点。我们使用 `agent`。
    # 这意味着这些是调用“代理”节点后获取的边缘。
    "agent",
    # 接下来，我们传入将确定下一个调用哪个节点的函数。
    should_continue,
    # 最后我们传入一个映射。
    # 键是字符串，而值是其他节点。
    # END是一个特殊节点，标记图形应该结束。
    # 我们将调用 `should_continue`，然后它的输出将是。
    # 将与此映射中的键进行匹配。
    # 根据匹配的内容，将调用该节点。
    {
        # 如果是`tools`，那么我们称这个工具为节点。
        "continue": "action",
        # 否则我们就完了。
        "end": END,
    },
)

# 我们现在从“工具”添加一条普通边到“代理”。
# 这意味着在调用`tools`之后，接下来将调用`agent`节点。
workflow.add_edge("action", "agent")

# 最后，我们将其汇编起来！
# 这将其编译成一个 LangChain 可运行对象。
# 这意味着你可以像使用任何其他可运行程序一样使用它。
app = workflow.compile()

from IPython.display import Image, display

display(Image(app.get_graph().draw_mermaid_png()))

from langchain_core.messages import HumanMessage

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
await app.ainvoke(inputs)

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
async for output in app.astream(inputs, stream_mode="updates"):
    # stream_mode="updates" yields dictionaries with output keyed by node name
    for key, value in output.items():
        print(f"Output from node '{key}':")
        print("---")
        print(value["messages"][-1].pretty_print())
    print("\n---\n")

inputs = {"messages": [HumanMessage(content="what is the weather in sf")]}
async for output in app.astream_log(inputs, include_types=["llm"]):
    # astream_log() 以 JSONPatch 格式生成请求的日志（这里是 LLMs）。
    for op in output.ops:
        if op["path"] == "/streamed_output/-":
            # 这是来自 .stream() 的输出。
            ...
        elif op["path"].startswith("/logs/") and op["path"].endswith(
            "/streamed_output/-"
        ):
            # 因为我们选择只包含LLM，因此这些是LLM令牌。
            try:
                content = op["value"].content[0]
                if "partial_json" in content:
                    print(content["partial_json"], end="|")
                elif "text" in content:
                    print(content["text"], end="|")
                else:
                    print(content, end="|")
            except:
                pass