提示生成从用户需求

在这个例子中，我们将创建一个帮助用户生成提示的聊天机器人。它首先从用户那里收集需求，然后生成提示（并根据用户输入进行优化）。这些步骤被分为两个独立的状态，LLM 决定何时在它们之间进行转换。

设置

首先，让我们安装所需的包并设置 OpenAI API 密钥（我们将使用的 LLM）。

python
%%capture --no-stderr
% pip install -U langgraph langchain_openai

import getpass
import os

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

_set_env("OPENAI_API_KEY")

设置 LangSmith 用于 LangGraph 开发

注册 LangSmith 以快速发现问题并提高 LangGraph 项目的性能。LangSmith 允许您使用跟踪数据来调试、测试和监控使用 LangGraph 构建的 LLM 应用程序——阅读更多关于如何开始的信息 这里。

收集信息

首先，让我们定义图的一部分，该部分将收集用户需求。这将是一个带有特定系统消息的 LLM 调用。它将访问一个工具，当它准备好生成提示时可以调用该工具。

python
from typing import List
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

template = """你的工作是从用户那里获取有关他们想要创建的提示模板类型的信息。

你应该从他们那里获取以下信息：

- 提示的目标是什么
- 哪些变量将传递到提示模板中
- 输出不应做什么的任何约束
- 输出必须遵守的任何要求

如果你无法辨别这些信息，请要求他们澄清！不要试图胡乱猜测。

在你能够辨别所有信息后，调用相关工具。"""

def get_messages_info(messages):
    return [SystemMessage(content=template)] + messages

class PromptInstructions(BaseModel):
    """关于如何提示 LLM 的说明。"""
    objective: str
    variables: List[str]
    constraints: List[str]
    requirements: List[str]

llm = ChatOpenAI(temperature=0)
llm_with_tool = llm.bind_tools([PromptInstructions])

def info_chain(state):
    messages = get_messages_info(state["messages"])
    response = llm_with_tool.invoke(messages)
    return {"messages": [response]}

生成提示

我们现在设置将生成提示的状态。这将需要一个单独的系统消息，以及一个过滤掉工具调用之前所有消息的函数（因为这是前一个状态决定生成提示的时候）。

python
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

# 新的系统提示
prompt_system = """根据以下要求，编写一个好的提示模板：

{reqs}"""

# 获取提示消息的函数
# 只会获取工具调用之后的消息
def get_prompt_messages(messages: list):
    tool_call = None
    other_msgs = []
    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls:
            tool_call = m.tool_calls[0]["args"]
        elif isinstance(m, ToolMessage):
            continue
        elif tool_call is not None:
            other_msgs.append(m)
    return [SystemMessage(content=prompt_system.format(reqs=tool_call))] + other_msgs

def prompt_gen_chain(state):
    messages = get_prompt_messages(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}

定义状态逻辑

这是聊天机器人所处状态的逻辑。如果最后一条消息是工具调用，那么我们处于“提示创建者”（prompt）应该响应的状态。否则，如果最后一条消息不是 HumanMessage，那么我们知道人类应该接下来响应，因此我们处于 END 状态。如果最后一条消息是 HumanMessage，那么如果之前有工具调用，我们处于提示状态。否则，我们处于“信息收集”（info）状态。

python
from typing import Literal
from langgraph.graph import END

def get_state(state):
    messages = state["messages"]
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "add_tool_message"
    elif not isinstance(messages[-1], HumanMessage):
        return END
    return "info"

创建图

我们现在可以创建图。我们将使用 SqliteSaver 来持久化对话历史。

python
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

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

memory = MemorySaver()
workflow = StateGraph(State)
workflow.add_node("info", info_chain)
workflow.add_node("prompt", prompt_gen_chain)

@workflow.add_node
def add_tool_message(state: State):
    return {
        "messages": [
            ToolMessage(
                content="Prompt generated!",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }

workflow.add_conditional_edges("info", get_state, ["add_tool_message", "info", END])
workflow.add_edge("add_tool_message", "prompt")
workflow.add_edge("prompt", END)
workflow.add_edge(START, "info")
graph = workflow.compile(checkpointer=memory)

使用图

我们现在可以使用创建的聊天机器人。

python
import uuid

cached_human_responses = ["hi!", "rag prompt", "1 rag, 2 none, 3 no, 4 no", "red", "q"]
cached_response_index = 0
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
while True:
    try:
        user = input("User (q/Q to quit): ")
    except:
        user = cached_human_responses[cached_response_index]
        cached_response_index += 1
    print(f"User (q/Q to quit): {user}")
    if user in {"q", "Q"}:
        print("AI: Byebye")
        break
    output = None
    for output in graph.stream(
        {"messages": [HumanMessage(content=user)]}, config=config, stream_mode="updates"
    ):
        last_message = next(iter(output.values()))["messages"][-1]
        last_message.pretty_print()

    if output and "prompt" in output:
        print("Done!")

完整代码

python
%%capture --no-stderr
% pip install -U langgraph langchain_openai

import getpass
import os

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

_set_env("OPENAI_API_KEY")

from typing import List
from langchain_core.messages import SystemMessage
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

template = """你的工作是从用户那里获取有关他们想要创建的提示模板类型的信息。

你应该从他们那里获取以下信息：

- 提示的目标是什么
- 哪些变量将传递到提示模板中
- 输出不应做什么的任何约束
- 输出必须遵守的任何要求

如果你无法辨别这些信息，请要求他们澄清！不要试图胡乱猜测。

在你能够辨别所有信息后，调用相关工具。"""

def get_messages_info(messages):
    return [SystemMessage(content=template)] + messages

class PromptInstructions(BaseModel):
    """关于如何提示 LLM 的说明。"""
    objective: str
    variables: List[str]
    constraints: List[str]
    requirements: List[str]

llm = ChatOpenAI(temperature=0)
llm_with_tool = llm.bind_tools([PromptInstructions])

def info_chain(state):
    messages = get_messages_info(state["messages"])
    response = llm_with_tool.invoke(messages)
    return {"messages": [response]}

from langchain_core.messages import AIMessage, HumanMessage, ToolMessage

# 新的系统提示
prompt_system = """根据以下要求，编写一个好的提示模板：

{reqs}"""

# 获取提示消息的函数
# 只会获取工具调用之后的消息
def get_prompt_messages(messages: list):
    tool_call = None
    other_msgs = []
    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls:
            tool_call = m.tool_calls[0]["args"]
        elif isinstance(m, ToolMessage):
            continue
        elif tool_call is not None:
            other_msgs.append(m)
    return [SystemMessage(content=prompt_system.format(reqs=tool_call))] + other_msgs

def prompt_gen_chain(state):
    messages = get_prompt_messages(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}

from typing import Literal
from langgraph.graph import END

def get_state(state):
    messages = state["messages"]
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "add_tool_message"
    elif not isinstance(messages[-1], HumanMessage):
        return END
    return "info"

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from typing import Annotated
from typing_extensions import TypedDict

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

memory = MemorySaver()
workflow = StateGraph(State)
workflow.add_node("info", info_chain)
workflow.add_node("prompt", prompt_gen_chain)

@workflow.add_node
def add_tool_message(state: State):
    return {
        "messages": [
            ToolMessage(
                content="Prompt generated!",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }

workflow.add_conditional_edges("info", get_state, ["add_tool_message", "info", END])
workflow.add_edge("add_tool_message", "prompt")
workflow.add_edge("prompt", END)
workflow.add_edge(START, "info")
graph = workflow.compile(checkpointer=memory)

import uuid

cached_human_responses = ["hi!", "rag prompt", "1 rag, 2 none, 3 no, 4 no", "red", "q"]
cached_response_index = 0
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
while True:
    try:
        user = input("User (q/Q to quit): ")
    except:
        user = cached_human_responses[cached_response_index]
        cached_response_index += 1
    print(f"User (q/Q to quit): {user}")
    if user in {"q", "Q"}:
        print("AI: Byebye")
        break
    output = None
    for output in graph.stream(
        {"messages": [HumanMessage(content=user)]}, config=config, stream_mode="updates"
    ):
        last_message = next(iter(output.values()))["messages"][-1]
        last_message.pretty_print()

    if output and "prompt" in output:
        print("Done!")

这个代码实现了一个帮助用户生成提示的聊天机器人，它首先收集用户的需求，然后根据这些需求生成提示，并根据用户的反馈进行优化。