AGI ｜一文快速上手LangChain的新利器：LangGraph！

前言

Part1 LLM Agent

（一）Agent概述

（二）Agent框架

Part2 LangGraph

（一）LangGraph介绍

（二）LangGraph组成

（三）LangGraph使用

（四）Demo演示

4.1 安装依赖

4.2 tools

4.3 Planner

4.4 Executor

4.5 Solver

4.6 定义Graph

4.7 运行

Part3 使用感受

前言

LLM Agent之所以受到广泛关注，是因为它代表了人工智能领域在处理和生成自然语言方面的最新进展，能够执行复杂的语言任务，如翻译、摘要、问答等，极大地推动了人机交互和自动化内容创作的边界。

而LangGraph提供了一种创新的方法来增强LLM的语义理解和生成能力，通过构建和利用语言的图谱结构，LangGraph有助于提高模型对语言细微差别的捕捉能力，从而在各种语言任务中实现更准确、更自然的表现。

本篇文章带大家一起，真实地上手尝试一下。

Part1 LLM Agent

（一）Agent概述

进入主题前还是想先聊下最近大家都在研究的Agent, 目前在LLM的应用场景上做的最多可能就是知识问答、内容总结、续写，但这肯定不能满足大家与LLM的期望，我们更希望它能帮我去完成一些复杂的工作，于是就有了了Agent的概念，让LLM充当Agent大脑的角色，理解用户的输入，将复杂问题分解成子任务，然后利用工具去一步步完成子任务，最后总结结果输出。

其中几个重要部分：

（1）规划 （效果取决大模型能力，目前能做的好一点的也就是GPT4了）：

子目标分解：agent将大任务拆分为更小的可管理的子目标，使得可以有效处理复杂任务。
反思与完善：agent对历史动作可以自我批评和自我反思，从错误中学习并在后续步骤里完善，从而改善最终结果的质量。

（2）记忆：

短期记忆：上下文学习即是利用模型的短期记忆学习
长期记忆：为agent提供保留和召回长期信息的能力，通常利用外部向量存储和检索实现

（3）工具使用 （目前很多开源的小模型都具有工具调用的能力，这里的工具不仅限API接口，是个广义的定义）

API: 接口
代码解释器：执行python、SQL
其他模型: 文生图
Agent：具有某种特定功能的Agent同样也能作为工具去使用

（二）Agent框架

目前比较热的几个框架，感兴趣的可以去了解下，这里不做介绍

X-Agent：国内的openBMB github.com/OpenBMB/X...
MetaGPT: meta github.com/geekan/Me...
AutoGPT: github.com/Significa...

虽然市面上现在存在很多Agent框架，但是大多体验下来其实还是都存在很多的问题，站在技术者的角度去看感觉离真正去落地到应用和业务场景上还是有不少的路要走的。

依赖模型：很多框架基本上都是要GPT3.5甚至GPT4才能跑出来，国内模型基本上效果都很差（不过现在Grok开源了），可以期待后面国内大模型能够有所改善吧
稳定性：对于复杂任务很难保证稳定性，对于要求高的业务场景来说基本没法上
消耗：商用模型token消耗过大，私有部署一个效果好点的千亿参数的模型似乎又不太可能，小点的效果又不太行
学习成本：目前市面长Agent框架太多，作为初学者和开发来说，目前没有办法能够说快速的就将某种框架能够应用到项目中去，很多框架应用起来还是相当复杂的，市面上很多对于框架的调研总结都是简单的测试，对于真正应用落地很难起到实质性的帮助。

Part2 LangGraph

（一）LangGraph介绍

LangGraph是Langchain新出的一个成员，是 LangChain 的 LangChain Expression Language （LCEL）的扩展。能够利用有向无环图的方式，去协调多个LLM或者状态，使用起来比 LCEL 会复杂，但是逻辑会更清晰。前期大家都学习了langchain,现在再上手学习langGraph就会容易许多，这也是我上面提到过的学习成本。我们可以把它也当做langchain扩展出来的Agent框架，langchain原有agent 的实现在LangGraph中都得到了重新实现，所以对于原来使用Langchain的系统去接入更容易。

（二）LangGraph组成

LangGraph主要就是图和节点：

nodes：图上流程中的每个节点，可以是工具调用或者模型
graph: 图可以当做运行的流程图，包含各个节点，节点间还可以加入流转条件
state: 作为节点间传递的上下文状态参数

ps: 这里node可以是任意的chain中runable的一种类, 就是说只要是使用langchain构建的chain是可以无缝链接的，例如LangGraph 可以直接使用CrewAI 当做node, 这样大大降低了以后扩展和接入其他框架的难度。

（三）LangGraph使用

下面介绍几种常用的使用案例：

AgentExecutor : 这个就是对应LangChain 以前实现的agents
ChatAgentExecutor：这是使用了OpenAI函数调用的Agent实现
Multi_Agent: 多代理的场景官方提供了三种:

1、Multi-agent collaboration: 如何创建两个Agent共同完成任务

2、Multi-agent with supervisor: 如何通过使用LLM作为中枢来协调各个Agent来分配工作

3、Hierarchical agent teams: 如何将Agent“团队”编排为可以协作解决问题的嵌套图

4、Planning Agent: 规划代理的场景官方也提供了三种:

Plan-and-execute: 这个应该算是我们应用中用到最多的场景，制定计划、执行任务、调用工具、结果反馈、更新计划。

Reasoning without Observation: 规划器生成一个任务列表，其观察结果被保存为变量。变量可以在后续任务中使用，以减少进一步重新规划的需要。相较于上面plan-and-execute 很大程度减少了token的消耗，比较推荐使用。

LLMCompiler: 这个和Rewoo是类似的做法，唯一区别是讲worker的任务执行变成并行处理，提升了处理速度，对于没有顺序要求的任务可以使用。

（四）Demo演示

下面我将开始使用LangGraph实现一个文档审查Agent的demo

4.1 安装依赖

requirements.txt文件：

langchain
langchain_openai
tavily-python
langgraph
langchainhub
unstructured[docx]
pymilvus
docx2pdf
python-multipart

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple/

4.2 tools

定义工具，我们使用Rewoo的模式去设计，那其实就只存在一个工具执行器的概念，也就是下面会提到的Executor，也是为了减少大模型交互的次数。这里我们只用提供我们的业务函数即可。

def handle_review_file(file_path):
    print('处理文件切片向量化存储')

def check_requirements(collection_name,query,check_req):
    print('校验文本内容要求')

def check_doc_format(standard_file_path,review_file_path): 
    print('校验文本格式要求')
    
def chat_model(query,check_req,content):    
    print('LLM chat')

4.3 Planner

我们可以参照rewoo中plannner的prompt格式来定义我们的prompt

PLAN_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) handle_file[input]: 处理文档的工具. 当你需要将文档处理文档的时候可以使用该工具.
(3) check_format[input]: 检查文档目录及格式的工具. 当你需要检查文档的格式、目录、大纲内容是否符合要求的时候可以使用该工具.
(4) check_requirement[input]: 检查文档内容是否符合要求. 当你需要检查文档某个检查点(check_point)内容是否符合标准要求的时候可以使用该工具.
(5) export_report[input]: 导出报告. 当你需要导出报告的时候可以使用该工具. 输入参数(报告内容).

For example,
Task: 请帮我检查一份文档,检查点包括：文档目录格式、入学要求、修学年限,最后将内容汇总后,导出一份完整的检查报告.
Plan: 处理文档. #E1 = handle_file['']
Plan: 检查文档目录格式. #E2 = check_format['']
Plan: 检查文档中入学要求是否符合要求. #E3 = check_requirement[入学要求,#E1]
Plan: 检查文档中修学年限是否符合要求. #E4 = check_requirement[修学年限,#E1]
Plan: 导出检查报告. #E5 = export_report[#E2,#E3,#E4]

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

Task: {task}"""

定义出Planner Node

llm = AzureChatOpenAI(
    azure_endpoint="https://xxxxxx.openai.azure.com/",
    azure_deployment="gpt-35-turbo",
    openai_api_version="2023-07-01-preview",
    openai_api_key="xxxxxxxxx",
    temperature=0
)
def get_plan(state: State):
    regex_pattern = r"\s*(#E\d+)\s*=\s*(\w+)\s*\[([^\]]+)\]"
    prompt_template = ChatPromptTemplate.from_messages([("user", PLAN_PROMPT)])
    planner = prompt_template | llm
    result = planner.invoke({"task": state["task"]['task_desc']})
    matches = re.findall(regex_pattern, result.content)
    return {"steps": matches, "plan_string": result.content}

定义出State

class Task(TypedDict):
    task_desc: str
    review_file_path: str
    standard_file_path: str

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

4.4 Executor

定义执行器的逻辑，用于执行各种任务

def _get_current_task(state: State):
    if state["results"] is None:
        return 1
    if len(state["results"]) == len(state["steps"]):
        return None
    else:
        return len(state["results"]) + 1

def tool_execution(state: State):
    _step = _get_current_task(state)
    step_name, tool, tool_input = state["steps"][_step - 1]
    _results = state["results"] or {}
    tool_input_list = [_results.get(k.strip(),k) for k in tool_input.split(",")]
    if tool == "handle_file":
        result = handle_review_file(state['task']['review_file_path'])
    elif tool == "check_format":
        result = check_doc_format(state['task']['standard_file_path'], state['task']['review_file_path'])
    elif tool == "check_requirement":
        result = check_requirements(tool_input_list[0], tool_input_list[1],state['task']['standard_file_path'])
    elif tool == "export_report":
        result = export_report(tool_input_list)
    else:
        raise ValueError
    _results[step_name] = str(result)
    return {"results": _results}

4.5 Solver

Solver用于接受Executor输出的结果并生成最终的响应

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: State):
    plan = ""
    for step_name, tool, tool_input in state["steps"]:
        _results = state["results"] or {}
        for k, v in _results.items():
            tool_input = tool_input.replace(k, v)
            step_name = step_name.replace(k, v)
        plan += f"Plan: \n{step_name} = {tool}[{tool_input}]"
    prompt = SOLVE_PROMPT.format(plan=plan, task=state["task"])
    result = llm.invoke(prompt)
    return {"result": result.content}

4.6 定义Graph

下面就是定义整个工作流程，将之前的planner、executor、solver作为节点添加进去

def _route(state):
    _step = _get_current_task(state)
    if _step is None:
        
        return "solve"
    else:
        
        return "tool"

graph = StateGraph(State)
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.set_entry_point("plan")
agent_app = graph.compile()

4.7 运行

task = Task(
    task_desc="请帮我检查一份文档,检查点包括：文档目录格式、入学要求、修学年限、课程说明和要求、毕业要求,最后将内容汇总后,导出一份完整的检查报告.",
    standard_file_path='./data/标准要求文档.docx',
    review_file_path='./data/待审核文档.docx'
)
for s in agent_app.stream({"task": task}, config={"recursion_limit": 100}):
    print(s)
    print("-----------------------------------------------------------")

这是生成的计划：

{'plan': {'steps': [('#E1', 'handle_file', "''"), ('#E2', 'check_format', '#E1'), ('#E3', 'check_requirement', '入学要求,#E1'), ('#E4', 'check_requirement', '修学年限,#E1'), ('#E5', 'check_requirement', '课程说明和要求,#E1'), ('#E6', 'check_requirement', '毕业要求,#E1'), ('#E7', 'export_report', '#E2,#E3,#E4,#E5,#E6')], 'plan_string': "Plan: 处理文档. #E1 = handle_file['']\nPlan: 检查文档目录格式. #E2 = check_format[#E1]\nPlan: 检查文档中入学要求是否符合要求. #E3 = check_requirement[入学要求,#E1]\nPlan: 检查文档中修学年限是否符合要求. #E4 = check_requirement[修学年限,#E1]\nPlan: 检查文档中课程说明和要求是否符合要求. #E5 = check_requirement[课程说明和要求,#E1]\nPlan: 检查文档中毕业要求是否符合要求. #E6 = check_requirement[毕业要求,#E1]\nPlan: 导出检查报告. #E7 = export_report[#E2,#E3,#E4,#E5,#E6]"}}
ps: 由于我只是模拟整个Agent执行的流程，里面工具输出的实际效果我并没有去关心。

ps: 由于我只是模拟整个Agent执行的流程，里面工具输出的实际效果我并没有去关心。