LangGraph 是 LangChain、LangServe 和 LangSmith 系列的最新成员，致力于使用法学硕士构建生成式 AI 应用程序。请记住，所有这些都是单独的软件包，必须单独进行 pip 安装。

在开始讨论 LangGraph 之前，我们需要先了解一下 LangChain 的两个主要概念。

1. 链：围绕 LLM 编写的用于执行任务的程序，例如自动 SQL 编写或 NER 提取链。请注意，链不能用于任何其他任务（甚至不能用于一般用例），并且如果尝试这样做可能会中断。链中要遵循的步骤是预先定义的并且不灵活。
2. 代理：链的一个更灵活的版本，代理通常是使用第三方工具（例如谷歌搜索，YouTube）启用的法学硕士，法学硕士本身决定下一步要做什么来解决给定的查询。

使用 LangGraph 改进 RAG

在本例中，我希望将 RAG 系统在数据库中的最终输出减少到 30 个字符以下。如果输出长度大于 30 个字符，我希望引入一个循环，使用不同的提示符再试一次，直到长度小于 30 个字符为止。这是用于演示的基本逻辑。你甚至可以实施复杂的逻辑来改善 RAG 结果。

我们要创建的图表如下

这里使用的版本包括：langchain===0.0.349、openai===1.3.8、langgraph===0.0.26。

首先，让我们导入重要内容并初始化 LLM。我使用的是 OpenAI API，但你也可以使用其他 LLM。

from typing import Dict, TypedDict, Optional
from langgraph.graph import StateGraph, END
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.embeddings.openai import OpenAIEmbeddings
llm = OpenAI(openai_api_key='your API')

接下来，我们将定义一个 StateGraph。

class GraphState(TypedDict):
    question: Optional[str] = None
    classification: Optional[str] = None
    response: Optional[str] = None
    length: Optional[int] = None
    greeting: Optional[str] = None
workflow = StateGraph(GraphState)

什么是状态图？

状态图是所有 LangGraph 流程的核心，它存储了我们在执行工作流时要存储的各种变量的状态。在本例中，我们有 5 个变量，这些变量的值将在执行图时更新，并与所有边和节点共享。

接下来，让我们从现有的矢量数据库中初始化一个 RAG 检索链。

def retriever_qa_creation():
        embeddings = OpenAIEmbeddings()
        db = Chroma(embedding_function=embeddings,persist_directory='/database',collection_name='details')
        qa = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=db.as_retriever())
        return qa
rag_chain = retriever_qa_creation()

接下来，我们将为该图添加节点

def classify(question):
    return llm("classify intent of given input as greeting or not_greeting. Output just the class.Input:{}".format(question)).strip()
def classify_input_node(state):
    question = state.get('question', '').strip()
    classification = classify(question) 
    return {"classification": classification}
def handle_greeting_node(state):
    return {"greeting": "Hello! How can I help you today?"}
def handle_RAG(state):
    question = state.get('question', '').strip()
    prompt = question
    if state.get("length")<30:
         search_result = rag_chain.run(prompt)
    else:
         search_result = rag_chain.run(prompt+'. Return total count only.')
    return {"response": search_result,"length":len(search_result)}

def bye(state):
    return{"greeting":"The graph has finished"}
workflow.add_node("classify_input", classify_input_node)
workflow.add_node("handle_greeting", handle_greeting_node)
workflow.add_node("handle_RAG", handle_RAG)
workflow.add_node("bye", bye)

接下来，我们将添加入口点和边缘

workflow.set_entry_point("classify_input")"classify_input")
workflow.add_edge('handle_greeting', END)
workflow.add_edge('bye', END)

在上述代码片段中：

• 我们在图中添加了一个入口点，即无论输入提示是什么，都要执行的第一个节点函数。
• 在这种情况下，如果我们的工作流程中出现 "handle_greeting"（处理问候）或 "bye"（再见），则图应该是END（终止工作流程的特殊节点）。

接下来，让我们添加条件边

def decide_next_node(state):
    return "handle_greeting" if state.get('classification') == "greeting" else "handle_RAG"
def check_RAG_length(state):
    return "handle_RAG" if state.get("length")>30 else "bye"
workflow.add_conditional_edges(
    "classify_input",
    decide_next_node,
    {
        "handle_greeting": "handle_greeting",
        "handle_RAG": "handle_RAG"
    }
)
workflow.add_conditional_edges(
    "handle_RAG",
    check_RAG_length,
    {
        "bye": "bye",
        "handle_RAG": "handle_RAG"
    }
)

条件边有助于根据条件（如 if-else）在 2 个节点之间做出选择。在创建的 2 条条件边中：

第 1 条条件边

遇到 "classifiy_input "时，根据 decide_next_node 函数的输出，选择 "handle_greeting "或 "handle_RAG"。

第 2 个条件边

如果遇到 "handle_RAG"，则根据 check_RAG_length 选择 "handle_RAG "或 "bye"。

编译并调用提示。初始时保持 length 变量=0

app = workflow.compile()compile()
app.invoke({'question':'Mehul developed which projects?','length':0})

#output 
{'question': 'Mehul developed which projects?',
 'classification': 'not_greeting',
 'response': ' 4',
 'length': 2,
 'greeting': 'The graph has finished'}

上述提示的图表流程如下所示

classify_input：情感将是 not_greeting
由于第一条条件边，转到句柄_RAG
由于长度=0，使用第一条提示并检索答案（总长度将大于 30）
由于第 2 个条件边，再次转到句柄_RAG
当长度>30 时，使用第 2 个提示符
由于第二个条件边，转到再见
结束

如果没有使用 LangGraph

rag_chain.run("Mehul developed which projects?")"Mehul developed which projects?")
#output 
"Mehul developed projects like ABC, XYZ, QWERTY. Not only these, he has major contribution in many other projects as well at OOO organization"

下一次输入

app.invoke({'question':'Hello bot','length':0})'question':'Hello bot','length':0})
#output
{'question': 'Hello bot',
 'classification': 'greeting',
 'response': None,
 'length': 0,
 'greeting': 'Hello! How can I help you today?'}

这里的流程会更简单

classify_input：情感将是 "问候"。
由于第 1 个条件边，转到 handle_greeting
结束

虽然我在这里应用的条件很简单，但这个框架可以通过添加更复杂的条件来轻松改善结果。