用FunctionCall实现文件解析(十)：接入LangGraph

发表于 更新于 分类于 阅读次数： 本文字数： 1k 阅读时长 ≈ 4 分钟

前言

既然我们在前面的$9$篇文章中做了这么多事情，接下来就再加一点新东西：LangGraph。

P.S.：虽然说官方最新版本已经更新到了比较后面，支持Runtime的版本，但是我的项目已经有点积重难返了，所以只能使用早些时候不支持Runtime的版本了。

P.S.：代码库已经开源至GitHub

状态转移

LangGraph的简介什么的我们就直接全部略过去吧，直接开始。

首先，我们知道，LangGraph将通过代码定义一张状态转移图，有明确的输入节点和明确的输出节点。而其中，LangGraph通信的核心功能就是其中的状态转移State和全局配置RunnableConfig。

首先，最常用的当然就是StateGraph，其中的参数并不是具体的实例，而是一个类，也就是传递状态转移信息的Schema。这个Schema定义了数据类型，甚至数据结构。

举个最简单的例子，我们定义一张状态转移图的时候，完全可以使用基本类型：

1
2
from langgraph.graph import StateGraph
graph = StateGraph(dict)

在之后传递状态转移过程的时候，就完全可以使用任何一种字典形式了。

同时，别忘了python中还有一个额外的typing库和tpying_extension库，其中可以使用typing.Dict和typing_extension.TypeDict来定义一个自定义的字典类型。

就比如：

1
2
3
4
from typing import List, Annotated
from typing_extension import TypeDict
class AppState(TypeDict):
    message: Annotated[List[str], reducer]

其中，还有一个神奇的东西，在LangGraph官方注释中，被定义为reducer，可以理解为任何List的累加器：

1
2
3
def reducer(b: list, a: int) -> list:
    if b: return b + [a]
    else: return [a]

这个reducer方法写得相当笼统。但实际上，我换一个写法，各位就很快能明白，具体会用在哪里：

1
2
3
4
5
from typing import List
from langchain_core.messages import BaseMessage
def reducer(b: List[BaseMessage], a: BaseMessage) -> list:
    if b: return b + [a]
    else: return [a]

没错，就是在消息列表中增加新的内容。

当然，LangGraph官方也为我们设计好了：

1
2
3
from langgraph.graph.message import add_messages
class AppState(TypeDict):
    message: Annotated[List[str], add_messages]

这样，我们就不需要自行实现了，只需要任由LangGraph自动完成即可。

为什么会自动完成？

相信不少人也会有这样的想法。

我们先来看看怎么添加节点：

1
2
3
4
5
6
def node_chat(state: AppState) -> Dict[str, BaseMessage]:
    try:
        response: AIMessage = llm.invoke(state.message)
        return {"message": response.content}
    except Exception as e:
        return {"message": str(e)}

也就是说，每一个节点本质上返回的内容也是严格与最开始我们定义的TypeDict类型的AppState中字段是严格一致的，返回的内容不可以超出AppState中定义的字段，但是可以省略未更新的部分。如果有更新，就放在字典中，然后StateGraph就能够自动管理状态转移AppState中的部分字段。

其中，Annotated修饰的字段可以根据reducer的定义自动拼接每一个节点返回的新内容，其他非可迭代类型也就能够通过自动覆盖实现硬更新。

更复杂的工程场景

在工程场景中，为了实现更为复杂的逻辑，我们的状态可以更丰富，比如我们可以定义很多状态属性：

1
2
3
4
5
class AppState:
    messages: Annotation[List[BaseMessage], reducer]    # 消息记录
    user_input: Union[BaseMessage, str]                 # 用户最新输入
    final_answer: Optional[str]                         # 最终的答案
    channel: int                                        # 意图分支

在节点上，可以做出如下动作：

  • 比如意图识别的节点，就可以仅返回{"channel": 0}，用来指示用户意图的分支，从而用conditional_egdes来定义意图分支上的不同策略。
  • 比如对话节点，就可以返回{"messages": [AIMessage(content='xxx')]}，从而实现聊天以及对话信息的长期自动维护。
  • 比如部分不去修改user_input的节点，就可以省略user_input参数的返回，后续节点访问过程中也将直接访问到当前未被修改的user_input参数，方便存取。

还有很多场景，都可以在以上的模式下进行灵活实现。