在本文中，你将学习如何构建一个面向任务的对话系统，帮助用户高质量地创建用户故事。该系统全部基于 LangGraph 的 “从用户需求生成提示 ”教程。

为什么需要使用 LangGraph？

在本教程中，我们假设你已经知道如何使用 LangChain。用户故事有一些组成部分，如目标、成功标准、执行计划和可交付成果。用户应提供其中的每一个组件，而我们需要 “手把手”逐一提供这些组件。如果只使用 LangChain，就需要大量的 “if ”和 “elses”。

有了 LangGraph，我们就可以使用图抽象来创建控制对话的循环。它还具有内置的持久性，因此我们无需担心主动跟踪图中发生的交互。

状态图（StateGraph）是 LangGraph 的主要抽象，用于创建图工作流。每个图都需要用状态模式（state_schema）进行初始化：状态模式是图中每个节点用来读写信息的模式类。

我们系统的流程将由多轮 LLM 和用户信息组成。主循环将包含以下步骤：

1. 用户说了什么
2. LLM 读取状态信息并决定是否准备好创建用户故事或用户是否应再次回复

我们的系统很简单，因此模式只包括对话中交换的信息。

from langgraph.graph.message import add_messages
class StateSchema(TypedDict):
    messages: Annotated[list, add_messages]

add_messages 方法用于将每个节点输出的消息合并到图状态中已有的消息列表中。

说到节点，LangGraph 的另两个主要概念是节点（Nodes）和边（Edges）。图中的每个节点都运行着一个函数，而每条边则控制着一个节点到另一个节点的流程。我们还有 START（开始）和 END（结束）虚拟节点，用于告诉图形在哪里开始执行，在哪里结束执行。

要运行系统，我们将使用 .stream() 方法。在我们构建并编译图形后，每一轮交互都将经过图形的 START（开始）到 END（结束），其路径（哪些节点应该运行或不运行）由我们的工作流结合图形的状态来控制。下面的代码是我们系统的主要流程：

config = {"configurable": {"thread_id": str(uuid.uuid4())}}
while True:
    user = input("User (q/Q to quit): ")
    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!")

在每次交互时（如果用户没有输入 “q ”或 “Q ”退出），我们使用 “更新 ”流模式（updates stream_mode）运行 graph.stream()，将用户的消息传递给用户，该模式会在图形的每一步之后流式传输状态更新。然后我们从 state_schema 中获取最后一条信息并打印出来。

在本文中，我们仍将学习如何创建图的节点和边，但首先让我们来谈谈 ToD 系统的总体架构，并学习如何使用 LLM、提示和工具调用来实现 ToD 系统。

ToD 系统的架构

构建端到端任务导向型对话系统框架的主要组成部分是：

1. 自然语言理解（NLU）：用于提取用户的意图和关键语段
2. 对话状态跟踪（DST）：用于跟踪用户在对话中的信念状态
3. 对话策略学习（DPL）：用于确定下一步行动
4. 自然语言生成（NLG）：用于生成对话系统响应

通过使用 LLM，我们可以将其中的一些组件合二为一。使用 LLM可以轻松实现 NLP 和 NLG 组件，因为理解和生成对话响应是它们的专长。

我们可以使用 LangChain 的系统消息（SystemMessage）来实现对话状态跟踪（DST）和对话策略学习（DPL），并在每次与 LLM 交互时始终传递该消息，从而为人工智能行为做好准备。每次与模型交互时，对话状态也应始终传递给 LLM。这意味着，我们要确保对话始终围绕我们希望用户完成的任务展开，始终告诉 LLM 对话的目标是什么以及它应该如何行动。为此，我们将首先使用提示符：

prompt_system_task = """Your job is to gather information from the user about the User Story they need to create.
You should obtain the following information from them:
- Objective: the goal of the user story. should be concrete enough to be developed in 2 weeks.
- Success criteria the sucess criteria of the user story
- Plan_of_execution: the plan of execution of the initiative
- Deliverables: the deliverables of the initiative
If you are not able to discern this info, ask them to clarify! Do not attempt to wildly guess. 
Whenever the user responds to one of the criteria, evaluate if it is detailed enough to be a criterion of a User Story. If not, ask questions to help the user better detail the criterion.
Do not overwhelm the user with too many questions at once; ask for the information you need in a way that they do not have to write much in each response. 
Always remind them that if they do not know how to answer something, you can help them.
After you are able to discern all the information, call the relevant tool."""

然后，每次向 LLM 发送信息时，我们都会添加这个提示：

def domain_state_tracker(messages):
    return [SystemMessage(content=prompt_system_task)] + messages

我们的 ToD 系统 LLM 实现的另一个重要概念是工具调用。如果你再次阅读 prompt_system_task 的最后一句话，就会发现 “在你能够识别所有信息后，请调用相关工具”。这样，我们就告诉 LLM，当它确定用户提供了所有的用户故事参数时，就应该调用工具来创建用户故事。我们的工具将使用带有用户故事参数的 Pydantic 模型来创建。

创建对话系统以构建用户故事

好了，是时候进行编码了。首先，我们要指定要使用的 LLM 模型，然后设置提示并绑定工具以生成用户故事：

import os
from dotenv import load_dotenv, find_dotenv
from langchain_openai import AzureChatOpenAI
from langchain_core.pydantic_v1 import BaseModel
from typing import List, Literal, Annotated
_ = load_dotenv(find_dotenv()) # read local .env file
llm = AzureChatOpenAI(azure_deployment=os.environ.get("AZURE_OPENAI_CHAT_DEPLOYMENT_NAME"),
                    openai_api_version="2023-09-01-preview",
                    openai_api_type="azure",
                    openai_api_key=os.environ.get('AZURE_OPENAI_API_KEY'),
                    azure_endpoint=os.environ.get('AZURE_OPENAI_ENDPOINT'),
                    temperature=0)
prompt_system_task = """Your job is to gather information from the user about the User Story they need to create.
You should obtain the following information from them:
- Objective: the goal of the user story. should be concrete enough to be developed in 2 weeks.
- Success criteria the sucess criteria of the user story
- Plan_of_execution: the plan of execution of the initiative
If you are not able to discern this info, ask them to clarify! Do not attempt to wildly guess. 
Whenever the user responds to one of the criteria, evaluate if it is detailed enough to be a criterion of a User Story. If not, ask questions to help the user better detail the criterion.
Do not overwhelm the user with too many questions at once; ask for the information you need in a way that they do not have to write much in each response. 
Always remind them that if they do not know how to answer something, you can help them.
After you are able to discern all the information, call the relevant tool."""
class UserStoryCriteria(BaseModel):
    """Instructions on how to prompt the LLM."""
    objective: str
    success_criteria: str
    plan_of_execution: str
llm_with_tool = llm.bind_tools([UserStoryCriteria])

正如我们之前所说，我们的图的状态只包括交换的消息和一个标志，以了解用户故事是否已创建。让我们首先使用 StateGraph 和此模式创建图表：

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
class StateSchema(TypedDict):
    messages: Annotated[list, add_messages]
workflow = StateGraph(StateSchema)

下图显示了最终图形的结构：

在顶部，我们有一个 talk_too_user 节点。这个节点可以：

• 结束对话（转到 finalize_dialogue 节点）
• 决定等待用户输入（转到 END 节点）

由于主循环永远运行（while True），所以每次图形到达END 节点时，都会再次等待用户输入。当我们创建循环时，这一点将变得更加清晰。

让我们创建图的节点，从 talk_too_user 节点开始。这个节点需要跟踪任务（在所有对话过程中保持主提示），还需要保持信息交换，因为对话的状态就存储在这里。该状态还保存了用户故事中哪些参数已被填满或未使用消息。因此，该节点每次都应添加 SystemMessage，并附加来自 LLM 的新消息：

def domain_state_tracker(messages):
    return [SystemMessage(content=prompt_system_task)] + messages
def call_llm(state: StateSchema):
    """
    talk_to_user node function, adds the prompt_system_task to the messages,
    calls the LLM and returns the response
    """
    messages = domain_state_tracker(state["messages"])
    response = llm_with_tool.invoke(messages)
    return {"messages": [response]}

现在，我们可以将 talk_too_user 节点添加到图中。我们将为它命名，然后传递我们创建的函数：

workflow.add_node("talk_to_user", call_llm)

这个节点应该是图中第一个运行的节点，因此我们用一条边来指定它：

workflow.add_edge(START, "talk_to_user")

到目前为止，图表是这样的：

为了控制图的流程，我们还将使用 LangChain 中的消息类。我们有四种类型的消息：

• SystemMessage：用于引导人工智能行为的消息
• HumanMessage：来自人类的信息
• AIMessage：聊天模型响应提示时返回的信息
• ToolMessage：包含工具调用结果的消息，用于将执行工具的结果传回模型

我们将使用图状态中最后一条信息的类型来控制 talk_too_user 节点上的流程。如果最后一条消息是 AIMessage，并且带有 tool_calls 关键字，那么我们就会转到 finalize_dialogue 节点，因为是时候创建用户故事了。否则，我们应该转到END 节点，因为现在是用户回答的时候了，所以我们要重新开始循环。

finalize_dialogue 节点应创建工具消息（ToolMessage），以便将结果传递给模型。tool_call_id 字段用于关联工具调用请求和工具调用响应。让我们创建这个节点并将其添加到图中：

def finalize_dialogue(state: StateSchema):
    """
    Add a tool message to the history so the graph can see that it`s time to create the user story
    """
    return {
        "messages": [
            ToolMessage(
                content="Prompt generated!",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        ]
    }
workflow.add_node("finalize_dialogue", finalize_dialogue)

现在让我们创建最后一个节点，即 create_user_story 节点。这个节点将使用创建用户故事的提示和对话中收集到的信息来调用 LLM。如果模型认为是时候调用工具了，那么 tool_calls 关键字的值就应该包含创建用户故事的所有信息。

prompt_generate_user_story = """Based on the following requirements, write a good user story:
{reqs}"""
def build_prompt_to_generate_user_story(messages: list):
    tool_call = None
    other_msgs = []
    for m in messages:
        if isinstance(m, AIMessage) and m.tool_calls: #tool_calls is from the OpenAI API
            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_generate_user_story.format(reqs=tool_call))] + other_msgs

def call_model_to_generate_user_story(state):
    messages = build_prompt_to_generate_user_story(state["messages"])
    response = llm.invoke(messages)
    return {"messages": [response]}
workflow.add_node("create_user_story", call_model_to_generate_user_story)

所有节点都已创建，现在是添加边的时候了。我们将为 talk_too_user 节点添加一条条件边。请记住，这个节点可以：

• 如果到了调用工具的时间，则结束对话（转到 finalize_dialogue 节点）
• 决定我们需要收集用户输入（转到结束节点）

这意味着，我们只检查最后一条信息是否是 AIMessage，并带有 tool_calls 关键字；否则，我们就应该转到 END 节点。让我们创建一个函数来检查这一点，并将其添加为一条边：

def define_next_action(state) -> Literal["finalize_dialogue", END]:
    messages = state["messages"]
    if isinstance(messages[-1], AIMessage) and messages[-1].tool_calls:
        return "finalize_dialogue"
    else:
        return END
workflow.add_conditional_edges("talk_to_user", define_next_action)

现在我们来添加其他边：：

workflow.add_edge("finalize_dialogue", "create_user_story")
workflow.add_edge("create_user_story", END)

这样，图形工作流程就完成了。是时候编译图形并创建循环来运行它了：

memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)
config = {"configurable": {"thread_id": str(uuid.uuid4())}}
while True:
    user = input("User (q/Q to quit): ")
    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 "create_user_story" in output:
        print("User story created!")

最后，让我们来测试一下这个系统：

结论

有了 LangGraph 和 LangChain，我们就能通过结构化的交互方式来构建引导用户的系统，通过使用 LLM 来帮助我们控制条件逻辑，从而降低创建系统的复杂性。

结合提示、内存管理和工具调用，我们可以创建直观有效的对话系统，为用户交互和任务自动化开辟新的可能性。