手撕LangChain代码，新手请进-六虎

假设你和我一样在预备24年的春招，在前端全栈外，再预备一些AI的内容是非常有必要的。24年是AI红利年，AIGC+各种岗位大厂机会会多些，赞同的请点赞。也欢迎朋友们加我微信shunwuyu, 一起交流。

前言

在上篇文章LangGraph 进化LangChain到多署理运转时 – (juejin.cn)中，咱们编写了一个组合search engine和twitter writter engine的例子。LangChain新推出的中心机制LangGraph,以多署理的办法，把活给干完了。今天，咱们将详尽学习使用到的代码。请翻开LangChain-Tutorials/langgraph_nodes_edges.ipynb at main sugarforever/LangChain-Tutorials (github.com)

代码

咱们将在colab调试咱们的代码，colab是Google推出的在线机器学习平台，除了一些模型练习、比较费GPU的使命外，我的学习型demo都能够在这儿free跑起来。

安装依靠

!pip install -q -U langchain langchain_openai langgraph google-search-results

首要!是什么意思呢？pip install 是安装依靠包，咱们在写python的时分，如同前面没有加过!。这儿的!是指在特定交互环境中（如 Jupyter Notebook、JupyterLab 或 Google Colab）的命令执行标记。用于告知该环境应当以体系shell来执行紧跟在这以后的命令，而不是将其当作Python代码来执行。
-q -U 是什么意思

-q 只显现错误信息，如果没有，就不显现，让安装以安静形式运转 -U update 表示安装最新版本的包
langchain_openai

咱们在玩node的时分，会到npmjs.org去检查包的详细信息。当咱们用pip安装包的时分，能够到pypi.org/，去看。

langchain框架负责中心模块，langchain_openai为LangChain集成了 OpenAI SDK的调用。

langgraph是新功能，为langchain带来了多署理的编程办法 google-search-results 来自serapi
注册serapi，并拿到sdk

咱们在开发生成式应用时，屡次使用到serapi先从google查找中拿到数据，再去生成。

环境变量设置

import os # os 模块供给了与操作体系相关的接口， 比方文件、目录，环境变量、进程办理等 这儿首要使用环境变量
from google.colab import userdata 
os.environ['SERPAPI_API_KEY'] = userdata.get('GOOGLE_API_KEY') os.environ['OPENAI_API_KEY'] = userdata.get('OPENAI_API_KEY') os.environ["LANGCHAIN_TRACING_V2"] = "true" 
# 项目名
os.environ["LANGCHAIN_PROJECT"] = "LangGraph" 
# LANGCHAIN_API_KEY
os.environ["LANGCHAIN_API_KEY"] = userdata.get('LANGSMITH_API_KEY')

colab 为咱们供给了userdata, 统一存放OPENAI_API_KEY、GOOGLE_API_KEY等,咱们能够在任一notebook里根据key取出userData里的值。

LangSmith

是一个用于构建出产级 LLM 应用程序的平台,用于交付,能够来到LangSmith注册拿到key, 它的使用咱们后续开新的文章再细讲。
查找试运转

from langchain_community.utilities import SerpAPIWrapper
search = SerpAPIWrapper() 
search.run("Obama's first name?")

langchain_community是langchain和一些第三方常用库结合的模块。当咱们安装了langchain、google-search-results后，咱们就能够调用langchain_community.utilities供给的SerpAPIWrapper来做查找作业。能够这样了解，langchain_community.utilities将第三方东西封装得更适合langchain业务，比方在上面的代码里，实例化后的search是能够run的。所以，咱们有langchain-opnai, 有SerpAPIWrapper, 第三方东西库由langchain_community.utilities供给。

Barack Hussein Obama II

引进相应的模块

import functools, operator, requests, os, json
from langchain.agents import AgentExecutor, create_openai_tools_agent 
from langchain_core.messages import BaseMessage, HumanMessage 
from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser 
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder 
from langgraph.graph import StateGraph, END 
from langchain.tools import tool 
from langchain_openai import ChatOpenAI 
from typing import Annotated, Any, Dict, List, Optional, Sequence, TypedDict

首要，咱们在规范库中引进functools, 它供给了一些函数式编程东西，比方等下要用到的装修器；operator包括了很多预界说的函数，如算术运算add等，高阶函数map等；requests用于发送恳求。

接着,咱们从agents模块引进 AgentExecutor、create_openai_tools_agent。AgentExecutor在LangChain框架中扮演着中心人物，它负责执行和办理不同类型的Agent, 并协调它们之间的交互。create_openai_tools_agent用于创立一个与OpenAI服务集成的Agent。让咱们以Agent的办法调用OpenAI。

紧接着，langchain_core供给了相应人物message类，这儿引进了BaseMessage和 HumanMessage。BaseMessage是基类，界说了音讯的基本结构和行为，HumanMessage能够用来表示用户的chat。

再者，from langchain.output_parsers.openai_functions import JsonOutputFunctionsParser 输出格局化咱们使用了JsonOutputFunctionsParser。LLM回来的是文本，output_parsers格局器供给了openai_functions,JsonOutputFunctionsParser函数会将文本转成json格局的输出。

又有 from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder langchain_core的prompts模块供给了MessagesPlaceholder和ChatPromptTemplate,用于Prompt设计。

接下来， from langgraph.graph import StateGraph, END LangGraph进场了，它是LangChain新功能，让咱们从graph里引进StateGraph和完毕结点

最终 from langchain_openai import ChatOpenAI在 LangChain 框架中引进一个与 OpenAI 的谈天模型，再后边是一堆的类型维束引进。

实例化LLM

llm = ChatOpenAI(model="gpt-4-turbo-preview")

咱们也能够使用gpt-3.5-turbo, 但是在处理一些复杂的agent时，gpt-4体现更优异。

界说查找函数

from langchain_core.messages import ( AIMessage, BaseMessage, ChatMessage, FunctionMessage, HumanMessage, SystemMessage )
@tool("web_search") 
def web_search(query: str) -> str: 
    """Search with Google SERP API by a query""" 
    search = SerpAPIWrapper() 
    return search.run(query)

Message 的类型真多啊，这个交给下次开篇讲，这儿咱们focus等下生成agent的web_search函数。functools供给的装修器形式，将web_search装修成为能够查找的东西函数。这个函数负责调用SerpAPIWrapper,根据传入的查询参数，拿到google的查找成果。

twitter_writer 东西函数

@tool("twitter_writer") 
def write_tweet(content: str) -> str: 
"""Based a piece of content, write a tweet.""" 
    chat = ChatOpenAI() 
    messages = [ 
    SystemMessage( content="You are a Twitter account operator." 
    " You are responsible for writing a tweet based on the content given." 
    " You should follow the Twitter policy and make sure each tweet has no more than 140 characters." ), 
    HumanMessage( content=content ), ]
    response = chat(messages) 
    return response.content

装修器仍然是协助函数成为创立agent的东西函数。函数通过ChatOpenAI供给的谈天实例，然后预备好音讯数组。SystemMessage 是谈天开端前的体系设置,在这儿指定了它的人物是Twitter writter, 使命是根据给定使命写一篇tweet, 要求不超过140字。HummanMessage 即用户谈天信息，content是传进来的内容，接着 response = chat(messages) 让谈天模型运转，并回来成果。这个东西函数首要负责生成。从这儿能够看出，twitter的生成是以谈天模型调用的办法生成。

GraphState

class AgentState(TypedDict):
# The annotation tells the graph that new messages will always 
# be added to the current states 
    messages: Annotated[Sequence[BaseMessage], operator.add] 
# The 'next' field indicates where to route to next 
    next: str

langgraph的第一个中心是graphstate, 这个状况标记当时的应用状况。 typing是 Python 的一个规范库模块，用于供给Type Hints功能。AgentState是一个根据TypedDict的类，里边包括相应的字段。

创立一个agent

def create_agent(llm: ChatOpenAI, tools: list, system_prompt: str):
    prompt = ChatPromptTemplate.from_messages( [ ( "system", system_prompt, ), 
        MessagesPlaceholder(variable_name="messages"),            MessagesPlaceholder(variable_name="agent_scratchpad"), ] ) 
    agent = create_openai_tools_agent(llm, tools, prompt) executor = AgentExecutor(agent=agent, tools=tools) 
    return executor 
def agent_node(state, agent, name): 
    result = agent.invoke(state) return {"messages": [HumanMessage(content=result["output"], name=name)]}

封装了一个create_agent办法，参数为llm、tools、system_prompt。咱们这儿要做的是一个谈天agent。咱们预备了prompt, 调用ChatPromptTemplate的from_messages办法。调用create_openai_tools_agent办法生成谈天agent,接受了参数分别是llm,tools数组， prompt。回来AgentExectutor的实例。

supervisor_chain

# 两个agent
members = ["Search_Engine", "Twitter_Writer"]

system_prompt = (
"You are a supervisor tasked with managing a conversation between the" 
" following workers: {members}. Given the following user request," 
" respond with the worker to act next. Each worker will perform a"
" task and respond with their results and status. When finished," 
" respond with FINISH." )

在这个体系prompt里，咱们首要指定它是一个监管上面两个agent的supervisor署理。

options = ["FINISH"] + members

function_def = { "name": "route", "description": "Select the next role.", "parameters": { "title": "routeSchema", "type": "object", "properties": { "next": { "title": "Next", "anyOf": [ {"enum": options}, ], } }, "required": ["next"], }, }

prompt = ChatPromptTemplate.from_messages( [ ("system", system_prompt), MessagesPlaceholder(variable_name="messages"), ( "system", "Given the conversation above, who should act next?" " Or should we FINISH? Select one of: {options}", ), ] ).partial(options=str(options), members=", ".join(members))
supervisor_chain = ( prompt | llm.bind_functions(functions=[function_def], function_call="route") | JsonOutputFunctionsParser() )

生成相应的agent和节点

# 调用create_agent办法，tool 为web_search, prompt 给的身份是web 查找引擎
search_engine_agent = create_agent(llm, [web_search], "You are a web search engine.") 
# 查找引擎节点 
functools.partial会回来一个对象，它预填了agent_node函数的部分参数，agent为 search_engine_agent, 名字为Search_Engine
search_engine_node = functools.partial(agent_node, agent=search_engine_agent, name="Search_Engine") 
# twitter 写作agent
twitter_operator_agent = create_agent(llm, [write_tweet], "You are responsible for writing a tweet based on the content given.") 
# twitter写作节点
twitter_operator_node = functools.partial(agent_node, agent=twitter_operator_agent, name="Twitter_Writer") 
# 实例化StateGraph， 将上面的AgentState 作为状况
workflow = StateGraph(AgentState) 
# 给作业流增加作业节点Search_Engine和Twitter_Writer
workflow.add_node("Search_Engine", search_engine_node) workflow.add_node("Twitter_Writer", twitter_operator_node) 
# 最终再增加监督节点，它会看在每个作业节点的状况
workflow.add_node("supervisor", supervisor_chain)

增加边

for member in members:
    workflow.add_edge(member, "supervisor") 
    # 条件边
    conditional_map = {k: k for k in members} 
    # 是否完毕
    conditional_map["FINISH"] = END 
    # 给监管节点增加条件边 加上supervisor上，它来决定
    # lambda x: x["next"]  去看在节点的next 值 是多少，下一步做什么
    workflow.add_conditional_edges("supervisor", lambda x: x["next"], conditional_map) 
    # 将supervisor 增加为初始点
    workflow.set_entry_point("supervisor") 
    # compile 编译生成graph
    graph = workflow.compile()

增加完后如图

给使命，并监听状况

# graph.stream是一个流式接口，遍历所有的节点，拿到输出信息
for s in graph.stream( 
    { 
        "messages": [ 
        # 使用HumanMesasage 代表用户宣布使命
            HumanMessage(content="Write a tweet about LangChain news") ]
    } 
): 
    # 如果没有__end__ 就输出反应
    if "__end__" not in s: 
        print(s) 
        print("----")

总结

静下心来，剖析了一版LangChain关于LangGraph的代码，对Prompt、Agent的了解更深入了。

LangGraph仍是不如AutoGen 谈天式多署理好了解，署理间的谈天是用chatOpenAI处理的，署理间的执行顺序是由边和条件边来处理的，而使命状况由grapState来搞定。

参考资料

LangChain-Tutorials/langgraph_nodes_edges.ipynb at main sugarforever/LangChain-Tutorials (github.com)

手撕LangChain代码，新手请进

前言

代码

安装依靠

总结

参考资料

相关文章

100+ ChatGPT-4 网络营销场景下的范式会话模板用例

BroadcastReceiver源码解析（Android12）

SwiftUI极简教程21:CoreData数据持久化框架的使用（中）

搭建交换机模拟环境及SSH连接，华为NSP软件入门使用教程

作者信息