LangGraph 如何工作？

让我们宏观地看看构成 LangGraph 的不同关键组件以及这些组件如何协同工作。

首先，什么是图？在计算机科学中，图指的是一种表示对象之间关系的数据结构。LangGraph 使用这个图的概念来组织 AI 代理及其交互。

LangGraph 状态

上图中我们没有看到但在 LangGraph 中是一个核心元素的是状态对象。状态对象保存应用程序内的上下文，它可以根据应用需求存储诸如对话消息和变量等值。

为了在信息从一个节点流向另一个节点时维持上下文，State会在图中的每个节点之间传递。每个节点在完成处理后都会返回一个更新后的State。

以下是我们如何定义应用程序State的示例：

...

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


class State(TypedDict):
    messages: Annotated[list, add_messages]


graph_builder = StateGraph(State)

这里我们的State接收一个消息列表以维护对话历史，确保节点之间保持上下文。add_messages内置函数决定了当新数据添加到State时列表应该如何更新。在这种情况下，add_messages将消息追加到列表中。

然后我们可以通过将我们的State传递给StateGraph类来构建我们的Graph，这样所有图节点都通过读写共享状态来通信。

LangGraph 节点

节点被定义为可以执行一组操作的 Python 函数。例如，一个节点可以与大型语言模型集成、处理信息、调用外部 API 或执行任何其他任务。

LangGraph 节点将图的状态作为参数，并在执行后返回更新后的状态。

以下是一个基本节点的示例，一个 Python 函数：

from langchain_openai import ChatOpenAI

...

llm = ChatOpenAI(model_name="gpt-4o")

def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

这是另一个使用大型语言模型的聊天机器人节点的示例：

LangGraph 中的边

两个节点之间的连接称为边。LangGraph 中的边可以是单向的（将一个节点连接到另一个节点，见上图中的节点 1 和 2）或双向的（见上图中的节点 3 和 4）。

LangGraph 中有三种不同类型的边：

1. 普通边：这些边很直接，它们直接将一个节点连接到另一个节点。
2. 条件边：这些边必须处理一个函数，以根据应用条件和需求确定下一个要去的节点。
3. START 和 END 边：这些特殊边定义了图初始化时调用哪个节点（START）以及哪些节点必须连接到 END 节点以终止图执行。

好的，我希望到目前为止一切都清楚了。让我们开始编码吧！

处理简历和工作数据

假设我们有两个方法，一个返回简历信息，另一个返回工作数据。为了简单起见，我们将通过以下方式模拟数据检索：

创建一个名为 modules 的新文件夹，并在其中创建一个新文件，命名为 resume.py，然后粘贴以下内容：

from typing import List, Optional
from pydantic import BaseModel, Field, validator

class WorkExperience(BaseModel):
    job_title: str = Field(description="工作职位或职位。")
    company: str = Field(description="公司名称。")
    experience: int = Field(description="在该工作中的经验年限。")
    responsibilities: List[str] = Field(description="工作职责列表。")

class Education(BaseModel):
    degree: str = Field(description="获得的学位。")
    school: str = Field(description="大学名称。")
    major: str = Field(description="主修科目。")
    year: Optional[int] = Field(description="毕业年份。")

    @validator('year', pre=True, always=True)
    def set_year(cls, v):
        if v is None:
            return 0
        return v

class Resume(BaseModel):
    """结构化简历数据。"""

    name: str = Field(description="个人姓名")
    professional_summary: str = Field(description="个人职业概述。")
    work_experience: List[WorkExperience] = Field(description="个人工作经验列表。")
    education: List[Education] = Field(description="个人教育资格列表。")
    skills: List[str] = Field(description="与工作相关的技能列表。")

    @classmethod
    def mock(cls):
        return cls(
            name='Jeff',
            professional_summary='创新型软件工程师，在科技行业拥有8年以上经验。公司X的高级开发人员，自由软件架构师，以及公司Y的初级开发人员。擅长开发可扩展应用程序，优化系统性能，并领导跨职能团队。精通英语和西班牙语。',
            work_experience=[
                WorkExperience(
                    job_title='高级开发人员',
                    company='公司X',
                    experience=5,
                    responsibilities=[
                        '领导可扩展Web应用程序的开发',
                        '优化系统性能并降低服务器成本',
                        '指导初级开发人员并进行代码审查',
                        '与产品经理合作定义项目需求',
                        '实施CI/CD管道以简化部署',
                        '为移动和Web应用程序开发RESTful API',
                        '确保应用程序安全并符合行业标准'
                    ]
                ),
                WorkExperience(
                    job_title='自由软件架构师',
                    company='独立顾问',
                    experience=2,
                    responsibilities=[
                        '为各种客户设计软件架构',
                        '提供技术咨询和项目管理',
                        '开发定制软件解决方案以满足客户需求',
                        '进行系统分析和性能调优',
                        '整合第三方服务和API',
                        '创建技术文档和用户手册'
                    ]
                ),
                WorkExperience(
                    job_title='初级开发人员',
                    company='公司Y',
                    experience=1,
                    responsibilities=[
                        '协助Web应用程序的开发',
                        '执行错误修复和代码维护',
                        '与高级开发人员合作完成项目任务',
                        '参与每日站会和冲刺规划',
                        '编写单元测试以确保代码质量',
                        '为开源项目做出贡献'
                    ]
                )
            ],
            education=[
                Education(
                    degree='计算机科学学士',
                    school='X大学',
                    major='计算机科学',
                    year=1999
                )
            ],
            skills=[
                '软件架构',
                '系统优化',
                '团队指导',
                '项目管理',
                'API开发',
                '持续集成/持续部署',
                '双语'
            ]
        )

上面的代码暴露了一个 mock() 方法，返回示例简历数据。在生产环境中，你可能需要读取 PDF 文件或抓取网站内容，然后解析输出，以使你的应用程序具有动态性。

我们将对工作数据做同样的处理，在 modules 文件夹内创建一个名为 job.py 的新文件，并粘贴以下内容：

from typing import List, Optional
from pydantic import BaseModel, Field

class Job(BaseModel):
    title: str = Field(description="工作职位或职位。")
    company: str = Field(description="公司名称。")
    location: Optional[str] = Field(description="工作地点。")
    salary: Optional[str] = Field(description="工作薪资范围。")
    description: str = Field(description="详细的工作描述。")
    responsibilities: List[str] = Field(description="工作职责列表。")
    benefits: Optional[List[str]] = Field(description="工作福利列表。")
    employment_type: Optional[str] = Field(description="雇佣类型（如全职、兼职）。")
    posted_date: Optional[str] = Field(description="工作发布日期。")

    @classmethod
    def mock(cls):
        return cls(
            title='软件工程师',
            company='科技公司',
            location='加利福尼亚州旧金山',
            salary='$100,000 - $120,000',
            description='我们正在寻找一位熟练的软件工程师加入我们的团队。',
            requirements=[
                '计算机科学或相关领域的学士学位',
                '3年以上软件开发经验',
                '精通Python和JavaScript',
                '有Django和React经验',
                '强大的问题解决能力'
            ],
            responsibilities=[
                '开发和维护Web应用程序',
                '与跨职能团队合作',
                '编写干净、可扩展和高效的代码',
                '参与代码审查',
                '排除故障和调试应用程序'
            ],
            benefits=[
                '健康保险',
                '401(k)匹配',
                '带薪休假',
                '灵活工作时间'
            ],
            employment_type='全职',
            posted_date='2024-10-01'
        )

很好，现在我们将包装这些模拟方法并将它们作为工具暴露出来，这些工具将成为我们 LangGraph 图中Tools节点的一部分。

为此，在你的主项目目录中创建一个名为 tools.py 的新文件，并粘贴以下代码：

from typing import Optional, Literal
from langchain_core.tools import tool

from modules.job import Job
from modules.resume import Resume

def process_job() -> Job:
    """处理工作数据。"""
    job = Job.mock()
    return job

def process_resume() -> Resume:
    """处理简历数据。"""
    resume = Resume.mock()
    return resume

@tool
def get_job(field: Optional[Literal['title', 'company', 'location', 'salary', 'description', 'responsibilities', 'benefits', 'employment_type', 'posted_date']] = None) -> str:
    """获取工作数据。"""
    job = process_job()
    if field:
        return getattr(job, field)
    return job.dict()

@tool
def get_resume(field: Optional[Literal['name', 'professional_summary', 'work_experience', 'education', 'skills']] = None) -> str:
    """获取简历数据。"""
    resume = process_resume()
    if field:
        return getattr(resume, field)
    return resume.dict()

我们创建了两个工具，get_job 和 get_resume，它们接受可选字段并从工作和简历数据集中返回完整数据或特定字段。使用结构化数据可能会提高整体响应质量，因为模型有具体的信息。

构建图

为了将所有内容整合在一起，我们需要创建我们的图，它由以下部分组成：

• 专家节点：该节点将与 OpenAI 的 GPT-4o 集成。
• 工具节点：该节点将检索简历和工作信息。

我们将使用 LangGraph 的 MessagesState 作为这个图的状态，因为我们只需要消息历史作为上下文。你可以在这里阅读更多关于 States 的信息：https://langchain-ai.github.io/langgraph/concepts/low_level/#state

在你的主目录中创建一个名为 agent.py 的新文件。以下是这个文件将包含的内容：

导入包

import os
from dotenv import load_dotenv

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import MemorySaver
from langchain_openai import ChatOpenAI
from tools import get_job, get_resume

导入了各种库和模块，用于处理类型、状态管理和与 OpenAI API 的交互。

加载工具和 LLM

tools = [get_job, get_resume]
llm = ChatOpenAI(
    model_name="gpt-4o",
    api_key=os.getenv("OPENAI_API_KEY")
).bind_tools(tools)

脚本定义了两个工具（get_job 和 get_resume）并将它们绑定到 OpenAI 的语言模型（LLM）。

这允许 LLM 在其操作过程中使用这些工具。别忘了在你的 .env 文件中添加你的 OPENAI_API_KEY。

创建我们的节点

这个函数定义了简历专家的核心逻辑。它确保代理只使用用户提供的信息，不编造任何细节。

def expert(state: MessagesState):
    system_message = """
        你是一位简历专家。你的任务是基于工作描述改进用户的简历。
        你可以使用提供的工具访问简历和工作数据。

        你绝不能提供用户没有的信息。
        这包括简历中没有的技能或经验。不要编造内容。
    """
    messages = state["messages"]
    response = llm.invoke([system_message] + messages)
    return {"messages": [response]}

tool_node = ToolNode(tools)

创建条件边

def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

这个函数决定一个节点是否必须 END（结束）或调用 tool（工具）来检索它没有的信息。在我们的情况下，这个条件边将连接到 Expert 节点，如下所示。

构建图

graph = StateGraph(MessagesState)

graph.add_node("expert", expert)
graph.add_node("tools", tool_node)

graph.add_edge(START, "expert")
graph.add_conditional_edges("expert", should_continue)
graph.add_edge("tools", "expert")

这里我们正在构建 LangGraph 工作流。

我们首先设置一个状态图来管理消息流。上面的 MessagesState 意味着所有节点消息将被追加到状态中。它定义了节点和边来控制代理如何处理用户输入并生成响应。

我们添加了节点、边（包括条件边，它查看 should_continue 来决定是否转到 tools 节点或 END）。

检查点和编译图

checkpointer = MemorySaver()

app = graph.compile(checkpointer=checkpointer)

应用循环

while True:
    user_input = input(">> ")
    if user_input.lower() in ["quit", "exit"]:
        print("Exiting...")
        break

    response = app.invoke(
        {"messages": [HumanMessage(content=user_input)]},
        config={"configurable": {"thread_id": 1}}
    )

    print(response["messages"][-1].content)

最后，代理处理每个输入并提供响应，交互式地帮助用户改进他们的简历。

结论

在这个教程中，我们看到了如何使用 LangGraph 构建一个超级简单的两节点代理。虽然官方文档可能有点复杂，但你看到实际上只需要几行代码就可以构建一个强大的应用。