前言

Reasoning without Observation的核心思想是，首先针对用户的目标提出一系列具体的操作，之后依次完成各个操作，期间不存在反思，执行完所有计划后总结得到结果。

一、LangGraph

1-1、介绍

LangGraph是一个专注于构建有状态、多角色应用程序的库，它利用大型语言模型（LLMs）来创建智能体和多智能体工作流。这个框架的核心优势体现在以下几个方面：

• 周期性支持：LangGraph允许开发者定义包含循环的流程，这对于大多数中智能体架构来说至关重要。这种能力使得LangGraph与基于有向无环图（DAG）的解决方案区分开来，因为它能够处理需要重复步骤或反馈循环的复杂任务。
• 高度可控性：LangGraph提供了对应用程序流程和状态的精细控制。这种精细控制对于创建行为可靠、符合预期的智能体至关重要，特别是在处理复杂或敏感的应用场景时。
• 持久性功能：LangGraph内置了持久性功能，这意味着智能体能够跨交互保持上下文和记忆。这对于实现长期任务的一致性和连续性非常关键。持久性还支持高级的人机交互，允许人类输入无缝集成到工作流程中，并使智能体能够通过记忆功能学习和适应。

1-2、特点

1. Cycles and Branching（循环和分支）

• 功能描述：允许在应用程序中实现循环和条件语句。
• 应用场景：适用于需要重复执行任务或根据不同条件执行不同操作的场景，如自动化决策流程、复杂业务逻辑处理等。

3. Persistence（持久性）

• 功能描述：自动在每个步骤后保存状态，可以在任何点暂停和恢复Graph执行，以支持错误恢复、等。
• 应用场景：对于需要中断和恢复的长时任务非常有用，例如数据分析任务、需要人工审核的流程等。

4. Human-in-the-Loop

• 功能描述：允许中断Graph的执行，以便人工批准或编辑Agent计划的下一步操作。
• 应用场景：在需要人工监督和干预的场合，如敏感操作审批、复杂决策支持等。

5. Streaming Support（流支持）

• 功能描述：支持在节点产生输出时实时流输出（包括Token流）。
• 应用场景：适用于需要实时数据处理和反馈的场景，如实时数据分析、在线聊天机器人等。

6. Integration with LangChain and LangSmith（与LangChain和LangSmith集成）

• 功能描述：LangGraph可以与LangChain和LangSmith无缝集成，但并不强制要求它们。
• 应用场景：增强LangChain和LangSmith的功能，提供更灵活的应用构建方式，特别是在需要复杂流程控制和状态管理的场合。

1-3、安装

pip install -U langgraph

1-4、什么是图？

图（Graph）是数学中的一个基本概念，它由点集合及连接这些点的边集合组成。图主要用于模拟各种实体之间的关系，如网络结构、社会关系、信息流等。以下是图的基本组成部分：

• 顶点（Vertex）：图中的基本单元，通常用来表示实体。在社交网络中，每个顶点可能代表一个人；在交通网络中，每个顶点可能代表一个城市或一个交通枢纽。
• 边（Edge）：连接两个顶点的线，表示顶点之间的关系或连接。边可以有方向（称为有向图），也可以没有方向（称为无向图）。
• 权重（Weight）：有时边会被赋予一个数值，称为权重，表示两个顶点之间关系的强度或某种度量，如距离、容量、成本等。

图可以根据边的性质分为以下几种：

• 无向图：边没有方向。
• 有向图：边有方向，通常用箭头表示。
• 简单图：没有重复的边和顶点自环（即边的两个端点是不同的顶点，且没有边从一个顶点出发又回到同一个顶点）。
• 多重图：可以有重复的边或顶点自环。
• 连通图：在无向图中，任意两个顶点之间都存在路径。

图在计算机科学中有广泛的应用，例如：

• 网络流问题：如最大流、最小割问题。
• 路径查找问题：如最短路径、所有路径问题。
• 社交网络分析：分析社交关系网，识别关键节点等。
• 推荐系统：通过分析用户之间的关系和偏好来推荐内容。

1-5、为什么选择图？

LangGraph之所以使用“图”这个概念，主要是因为图（Graph）在表达复杂关系和动态流程方面具有天然的优势。以下是使用图概念的一些具体原因：

• 表达复杂关系：在构建智能体应用时，各组件之间可能存在复杂的关系和交互。图结构可以很好地表示这些关系，包括节点（代表状态或操作）和边（代表转移或关系）。
• 动态流程管理：智能体在执行任务时，往往需要根据不同的输入或状态动态调整其行为。图结构允许灵活地表示这些动态流程，如循环、分支和并行路径。
• 可扩展性：图结构易于扩展。随着应用复杂度的增加，可以轻松地在图中添加新的节点和边，而不需要重写整个流程。
• 可视化：图的可视化特性使得开发者能够直观地理解和调试智能体的行为。通过图形化的表示，可以更快速地识别问题和优化点。
• 循环和递归：许多智能体应用需要处理循环或递归逻辑，如图结构可以自然地表示这种循环引用和重复过程。
• 灵活的控制流：与传统的线性流程（如有向无环图DAG）相比，图结构支持更复杂的控制流，包括条件分支和并发执行。
• 启发式算法和数据流：图算法（如最短路径、网络流等）可以为优化智能体行为提供启发，特别是在处理数据流和资源分配时。

在LangChain的简答链中无法实现的循环场景：

1、代码生成与自我纠正：

• 场景描述：利用LLM自动生成软件代码，并根据代码执行的结果进行自我反省和修正。
• LangGraph应用：LangGraph可以创建一个循环流程，首先生成代码，然后测试执行，根据执行结果反馈给LLM，让它重新生成或修正代码，直到达到预期的执行效果。这种循环机制在传统的链式（Chain）结构中难以实现。

2、Web自动化导航：

• 场景描述：自动在Web上进行导航，例如自动填写表单、点击按钮或从网站上抓取信息。
• LangGraph应用：LangGraph可以定义一个包含循环的流程，使得智能体能够在进入下一界面时，根据多模态模型的决定来执行不同的操作（如点击、滚动、输入等），直到完成特定任务。这种循环和条件逻辑的运用在LangGraph中得到了很好的支持。

总结来说：LangGraph可以表达更复杂的关系，更灵活，控制更精细，具备循环能力。

1-6、LangGraph应用的简单示例—CRAG（自我改正型RAG）

LangGraph： 是 LangChain 的扩展库，不是独立的框架。它能协调 Chain、Agent 和 Tool 等组件，支持 LLM 循环调用和 Agent 过程的精细化控制。LangGraph 使用状态图（StateGraph）代替了 AgentExecutor 的黑盒调用，通过定义图的节点和边来详细定义基于 LLM 的任务。在任务运行期间，它会维护一个中央状态对象，该对象会根据节点的变化不断更新，其属性可根据需要进行自定义。相比于 AgentExecutor，LangGraph 可以更加精细的进行控制：

CRAG： 顾名思义，一种RAG的变体，结合了对检索到的文档的自我反思/自我评分。

图表展示了一个查询处理流程，涉及多个阶段和决策点：

• Question（提问）：这是整个流程的开始点，用户提出一个问题。
• Retrieve Node（检索节点）：系统尝试从数据库或索引中检索与问题相关的信息。
• Grade Node（评分节点）：对检索到的信息进行评估，判断其相关性或准确性。
• Decision Point（决策点）：根据评分节点的输出，系统会做出是否继续当前路径还是选择替代路径的决定。
  如果没有发现任何无关的文档（“Any doc irrelevant”? “No”），则流程直接跳到“Answer（答案）”节点。
  如果发现了无关的文档（“Any doc irrelevant”? “Yes”），则进入下一个阶段。重查。
• Re-write Query Node（重写查询节点）：由于检索到的某些文档被认为是不相关的，系统会对原始查询进行重新表述，以便更准确地反映用户的需求。
• Web Search Node（网页搜索节点）：使用重写的查询在互联网上搜索更多信息。
• Answer（答案）：最终，系统将生成的答案返回给用户。

节点可以是可调用的函数、工具、Agent、或者是一个可运行的chain。

1-7、LangGraph基础概念

1-7-1、Graphs（图的概念&关键组件&如何构建）

在LangGraph框架中，“Graphs”（图）是核心概念之一，用于图形化 智能体（agents）的工作流程。（即将工作流程建模为图形），主要使用三个关键组件来定义：

• State: 状态，一个共享的数据结构。
• Nodes：节点，编辑Agent逻辑的python函数，接收当前状态作为输入，执行一系列计算后，返回更新的状态。
• Edges：边，基于当前状态决定下一个执行节点。边可以是条件分支，或者固定转换。

通过组合、拼接节点和边，可以创建复杂的工作流程。

Graphs 执行开始时，所有节点都以初始状态开始，当节点完成操作后，它会沿着一条或者多条边向其他节点发送消息，之后，接收方节点执行其函数，将生成的消息传递给下一组节点。直到没有消息传递！
简单说：节点完成操作，边决策下一步干什么。

参数：

• StateGraph：状态图，用于将用户定义的对象参数化。
• MessageGraph：消息图，除了聊天机器人外基本不使用。

构建图： 首先需要定义state，之后需要添加各个节点和边，最后就可以编译图了。（对图结构的一些基本检查，确保图没有孤立节点，另外还可以指定一些运行时的参数）。调用以下方法来编译图：

graph = graph_builder.compile(...)

1-7-2、State（状态）

State：

1. 定义： 状态是Graph中的一个共享数据结构，它存储了所有与当前执行上下文相关的信息。
2. 数据结构： 状态可以是任何Python类型，通常使用TypedDict或PydanticBaseModel。TypedDict是一个Python字典，它允许对字典键和值的类型进行注解。
3. 作用： 状态用于存储和传递应用程序的数据，使得节点可以基于这些数据执行操作。它提供了节点之间的通信机制，因为每个节点都可以读取前一个节点更新的状态，并在此基础上进行操作。
4. 管理： 状态的管理是自动的。当一个节点执行并返回一个更新后的状态时，LangGraph框架会确保这个新状态被传递到下一个节点。
5. 生命周期： 状态的生命周期与图的执行周期相匹配，它从图的初始状态开始，并在图的每个节点执行时更新，直到图执行结束。

1-7-3、Annotated（数据类型）

Annotated作用：

1. 元数据添加：Annotated允许开发者在类型提示中添加额外的信息，这些信息可以被类型检查器、框架或其他工具使用。
2. 类型提示增强：它提供了一种方式来增强现有的类型提示，而不需要创建新的类型。
3. 代码文档：Annotated可以作为一种文档形式，提供关于变量、函数参数或返回值的额外信息。

用法1： DistanceInCm是一个带注释的整数类型。注释 “Units: cm” 说明了这个整数代表的是以厘米为单位的距离。注释可以用来作为文档，说明变量的用途或期望的值。

from typing import Annotated
from typing_extensions import Annotated  # 如果标准库中没有Annotated

# 定义一个带注释的整数类型
# 这里的 "Units: cm" 是一个注释，它不会改变类型的行为
DistanceInCm = Annotated[int, "Units: cm"]

def measure_distance(distance: DistanceInCm) -> DistanceInCm:
    # 这里我们假设函数会测量距离，并返回以厘米为单位的距离
    # 注意：函数的实现并不关心注释 "Units: cm"
    return distance

# 使用带注释的类型
distance: DistanceInCm = 10  # 10 厘米
new_distance = measure_distance(distance)

用法2： messages 变量，其类型被注解为 Annotated[list, add_messages]。list 表示 messages 键的值应该是一个列表。add_messages 是一个函数，它在 Annotated 注解中使用，提供了关于如何更新状态字典中 messages 的额外信息。

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    # Messages have the type "list". The `add_messages` function
    # in the annotation defines how this state key should be updated
    # (in this case, it appends messages to the list, rather than overwriting them)
    messages: Annotated[list, add_messages]

1-7-4、Node（节点）

Node： 在LangGraph框架中，节点（Nodes）是Python函数，它们编码了智能体（agents）的逻辑。其中第一个位置参数是State（名称）。第二个位置参数是config（Node对应的处理逻辑）。

from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph

builder = StateGraph(dict)


def my_node(state: dict, config: RunnableConfig):
    print("In node: ", config["configurable"]["user_id"])
    return {"results": f"Hello, {state['input']}!"}


# The second argument is optional
def my_other_node(state: dict):
    return state


builder.add_node("my_node", my_node)
builder.add_node("other_node", my_other_node)
...

Start节点： 特殊节点，表示用户输入发送到Graph的节点。该节点的主要目的是确定首先应该调用哪些节点。

from langgraph.graph import START

graph.add_edge(START, "node_a")

End节点： 特殊节点，确定哪条边完成后，没有后续操作。

from langgraph.graph import END

graph.add_edge("node_a", END)

1-7-5、Edge（边）

在LangGraph框架中，边（Edges）是用于连接节点的对象，它们定义了节点之间的转换逻辑。每条边都连接两个节点：一个源节点和一个目标节点。边的主要功能是确定何时应该从源节点跳转到目标节点。

• Normal Edges：正常边，直接从一个节点转到下一个节点。
• Conditional Edges：调用一个函数以确定接下来要转到哪个节点。

# 正常边，直接从节点A跳转到节点B
graph.add_edge("node_a", "node_b")

# 条件边，从节点A选择性的跳转到下一条边，routing_function为跳转的逻辑方法。
graph.add_conditional_edges("node_a", routing_function)

1-7-6、Command

概念： Command可以很方便的既进行走向控制，又可以更新状态。个人理解，代码更加简洁，省去使用条件边的流程。

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    return Command(
        # state update
        update={"foo": "bar"},
        # control flow
        goto="my_other_node"
    )

动态控制： 类似条件边

def my_node(state: State) -> Command[Literal["my_other_node"]]:
    if state["foo"] == "bar":
        return Command(update={"foo": "baz"}, goto="my_other_node")

二、Reasoning without Observation

2-1、介绍

Reasoning without Observation：核心思想是，首先针对用户的目标提出一系列具体的操作，之后依次完成各个操作，期间不存在反思，执行完所有计划后总结得到结果，如上图所示。主要由三条链组成：

• Planner：生成具体的计划。（针对于用户任务的一系列具体执行步骤）
• Worker： 使用提供的参数执行对应的步骤。（按照执行步骤依次执行）
• Solver：总结使用工具执行的每一步结果，生成最终的答案。

生成类似以下内容的计划：

Plan: <reasoning>
#E1 = Tool[argument for tool]
Plan: <reasoning>
#E2 = Tool[argument for tool with #E1 variable substitution]
...

相比于React，更加简洁。而且避免消耗非常多的Token。

2-2、初始设置(模型配置&搜索引擎)

tavily 搜索API申请地址: https://docs.tavily.com/docs/rest-api/api-reference

pip install -U langchain-community tavily-python

TavilySearchResults参数介绍:

• max_results：最大返回搜索数量
• include_answer：是否包含答案
• include_images： 是否包含图片

简易Demo：

import os 

os.environ["TAVILY_API_KEY"] = ""

from langchain_community.tools import TavilySearchResults

tool = TavilySearchResults(
    max_results=5,
    include_answer=True,
    include_raw_content=True,
    include_images=True,
    # search_depth="advanced",
    # include_domains = []
    # exclude_domains = []
)
tools = [tool]
tool.invoke({'query': '谁是世界上最美丽的女人？'})

配置模型以及tavily 搜索：

from typing import List
from typing_extensions import TypedDict
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain import hub
from langchain_openai import ChatOpenAI
import os
from langgraph.prebuilt import create_react_agent

search = TavilySearchResults(max_results=5)

# 这里我们使用通义千问的模型
model = ChatOpenAI(
    model="qwen-max",
    temperature=0,
    max_tokens=1024,
    timeout=None,
    max_retries=2,
    api_key=os.environ.get('DASHSCOPE_API_KEY'),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)

2-3、定义图状态

注： 在LangGraph中，每个节点都共享图状态，图状态是所有节点在被调用时的输入，我们需要的状态字典包含：

• task：任务目标，即用户输入。
• plan_string：具体的实施步骤，由Planner制定。
• steps：上一步的结果，通过正则化得到更规范的实施步骤列表。
• results：每一个实施步骤得到的结果，字典。
• result：最终结果。

from typing import List
from typing_extensions import TypedDict


class ReWOO(TypedDict):
    task: str
    plan_string: str
    steps: List
    results: dict
    result: str

2-4、Planner

Planner： 生成一系列具体的计划，在与state交换信息时，会使用正则表达式提取出对应的参数。

注意： 在提示词中会提到使用的工具，如果需要加入其他工具，则更改以下提示词。

prompt = """For the following task, make plans that can solve the problem step by step. For each plan, indicate \
which external tool together with tool input to retrieve evidence. You can store the evidence into a \
variable #E that can be called by later tools. (Plan, #E1, Plan, #E2, Plan, ...)

Tools can be one of the following:
(1) Google[input]: Worker that searches results from Google. Useful when you need to find short
and succinct answers about a specific topic. The input should be a search query.
(2) LLM[input]: A pretrained LLM like yourself. Useful when you need to act with general
world knowledge and common sense. Prioritize it when you are confident in solving the problem
yourself. Input can be any instruction.

For example,
Task: Thomas, Toby, and Rebecca worked a total of 157 hours in one week. Thomas worked x
hours. Toby worked 10 hours less than twice what Thomas worked, and Rebecca worked 8 hours
less than Toby. How many hours did Rebecca work?
Plan: Given Thomas worked x hours, translate the problem into algebraic expressions and solve
with Wolfram Alpha. #E1 = WolframAlpha[Solve x + (2x − 10) + ((2x − 10) − 8) = 157]
Plan: Find out the number of hours Thomas worked. #E2 = LLM[What is x, given #E1]
Plan: Calculate the number of hours Rebecca worked. #E3 = Calculator[(2 ∗ #E2 − 10) − 8]

Begin! 
Describe your plans with rich details. Each Plan should be followed by only one #E.

Task: {task}"""

task = "what is the exact hometown of the 2024 mens australian open winner"
result = model.invoke(prompt.format(task=task))
print(result.content)

输出：
Plan: Identify the 2024 men’s Australian Open winner. #E1 = Google[2024 men’s Australian Open winner]
Plan: Once we have the name of the 2024 men’s Australian Open winner, find out their exact hometown. #E2 = Google[exact hometown of #E1]

2-5、构建Planner节点

get_plan: 构建Planner节点，该节点接受图状态，提取用户输入，生成详细的计划，并使用正则表达式去提取关键具体步骤并且更新图状态

import re

from langchain_core.prompts import ChatPromptTemplate

# Regex to match expressions of the form E#... = ...[...]
regex_pattern = r"Plan:\s*(.+)\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"
prompt_template = ChatPromptTemplate.from_messages([("user", prompt)])
planner = prompt_template | model


def get_plan(state: ReWOO):
    task = state["task"]
    result = planner.invoke({"task": task})
    # Find all matches in the sample text
    matches = re.findall(regex_pattern, result.content)
    return {"steps": matches, "plan_string": result.content}

2-6、Executor（执行者节点的构建）

get_current_task： 接受图状态，该函数用于确认计划具体执行到第几步。

tool_execution： Worker，执行节点，接受图状态，确定具体的执行步骤，并调用相应的工具。更新图状态，写入更新工具得到的结果。

def _get_current_task(state: ReWOO):
    # 如果state里不包含results，即结果字段，则返回1.
    # 即在后续先执行第0个计划。
    if "results" not in state or state["results"] is None:
        return 1
    # 检测是不是把每一步都执行过了
    if len(state["results"]) == len(state["steps"]):
        return None
    else:
        # 如果未执行完，则步数加1
        return len(state["results"]) + 1


def tool_execution(state: ReWOO):
    """Worker node that executes the tools of a given plan."""
    _step = _get_current_task(state)
    # 分别是计划，步名，使用到的工具，工具对应的输入
    _, step_name, tool, tool_input = state["steps"][_step - 1]
    # or {}：这是一个逻辑或操作，它的作用是在state["results"]为None或者为空字典{}时提供一个默认值。
    # 如果state["results"]是None或者空字典，那么整个表达式的结果将是后面的{}，也就是一个空字典。
    # if "results" in state:
    #     if state['results']:
    #         return state['results']
    #     else:
    #         return {}
    # else:
    #     return {}
    _results = (state["results"] or {}) if "results" in state else {}
    # 把已经得到的结果替换掉即将要输入工具的符号。
    for k, v in _results.items():
        tool_input = tool_input.replace(k, v)
    if tool == "Google":
        result = search.invoke(tool_input)
    elif tool == "LLM":
        result = model.invoke(tool_input)
    else:
        raise ValueError
    _results[step_name] = str(result)
    return {"results": _results}

2-7、Solver节点

solver: Solver节点接受图状态，获取到完整的执行计划以及每一步对应的结果。之后根据提示词以及目标，生成最终的结果。

solve_prompt = """Solve the following task or problem. To solve the problem, we have made step-by-step Plan and \
retrieved corresponding Evidence to each Plan. Use them with caution since long evidence might \
contain irrelevant information.

{plan}

Now solve the question or task according to provided Evidence above. Respond with the answer
directly with no extra words.

Task: {task}
Response:"""


def solve(state: ReWOO):
    plan = ""
    for _plan, step_name, tool, tool_input in state["steps"]:
        _results = (state["results"] or {}) if "results" in state else {}
        for k, v in _results.items():
            tool_input = tool_input.replace(k, v)
            step_name = step_name.replace(k, v)
        plan += f"Plan: {_plan}\n{step_name} = {tool}[{tool_input}]"
    prompt = solve_prompt.format(plan=plan, task=state["task"])
    result = model.invoke(prompt)
    return {"result": result.content}

2-8、Graph定义

route： 路由函数，用于执行过程中确认下一步跳转到solver节点还是 Worker节点。

def _route(state):
    _step = _get_current_task(state)
    # 看一下步数是否为空如果执行完，则跳到solve节点，如果没执行完，则接着到tool节点去执行
    if _step is None:
        # We have executed all tasks
        return "solve"
    else:
        # We are still executing tasks, loop back to the "tool" node
        return "tool"


from langgraph.graph import END, StateGraph, START

graph = StateGraph(ReWOO)
graph.add_node("plan", get_plan)
graph.add_node("tool", tool_execution)
graph.add_node("solve", solve)
graph.add_edge("plan", "tool")
graph.add_edge("solve", END)
graph.add_conditional_edges("tool", _route)
graph.add_edge(START, "plan")

app = graph.compile()
for s in app.stream({"task": task}):
    print(s)
    print("---")

# Print out the final result
print(s["solve"]["result"])

输出：

参考文章：
LlamaIndex 官方文档
langgraph官方教程
langgraph操作指南
langgraph概念指南
langgraph API 参考
langgraph 词汇表
langgraph 快速入门
彻底搞懂LangGraph【1】：构建复杂智能体应用的LangChain新利器
LangChain 79 LangGraph 从入门到精通一
Reasoning without Observation官方文档

总结

效果还可以，但是细节之处还需要进一步修改。😶‍🌫️