工作流与代理

本指南回顾了代理系统中的常见模式。在描述这些系统时，区分“工作流”和“代理”是有用的。一种解释这种差异的方法可以在Anthropic的《构建有效的代理》博客文章中找到：

工作流是通过预定义的代码路径来协调大型语言模型（LLMs）和工具的系统。 相比之下，代理系统则是由LLMs动态地指导自己的过程和工具使用，保持对完成任务方式的控制。

这里是一种简单的方式来可视化这些差异：

在构建代理和工作流时，LangGraph提供了一系列好处，包括持久性、流处理以及调试和支持部署的功能。

设置

您可以使用支持结构化输出和工具调用的任意聊天模型。以下展示了安装包、设置API密钥以及测试Anthropic结构化输出/工具调用的过程。

安装依赖项

pip install langchain_core langchain-anthropic langgraph 

初始化一个LLM

import os
import getpass

from langchain_anthropic import ChatAnthropic

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

llm = ChatAnthropic(model="claude-3-5-sonnet-latest")

构建模块：增强型LLM

LLM具有支持构建工作流和代理的增强功能。这些包括结构化输出和工具调用，如Anthropic博客上的“构建有效代理”一文中的图像所示：

# 结构化输出模式
from pydantic import BaseModel, Field

class SearchQuery(BaseModel):
    search_query: str = Field(None, description="优化网络搜索的查询。")
    justification: str = Field(
        None, description="为什么这个查询与用户的请求相关。"
    )


# 使用结构化输出模式增强LLM
structured_llm = llm.with_structured_output(SearchQuery)

# 调用增强后的LLM
output = structured_llm.invoke("钙CT评分与高胆固醇有何关系？")

# 定义一个工具
def multiply(a: int, b: int) -> int:
    return a * b

# 使用工具增强LLM
llm_with_tools = llm.bind_tools([multiply])

# 输入触发工具调用的信息并调用LLM
msg = llm_with_tools.invoke("2乘以3是多少？")

# 获取工具调用
msg.tool_calls

提示链式处理

在提示链式处理中，每个LLM调用都会处理前一个调用的输出。

正如Anthropic博客中的构建有效的代理程序所提到：

提示链式处理将任务分解为一系列步骤，其中每个LLM调用都会处理前一个调用的输出。您可以在任何中间步骤添加程序检查（见下图中的“门”）以确保过程仍在正确的轨道上。

何时使用此工作流：此工作流适用于可以轻松且干净地分解为固定子任务的情况。主要目标是通过使每个LLM调用成为更简单的任务来权衡延迟以获得更高的准确性。

图形API功能API

from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display


# 图状态
class State(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str


# 节点
def generate_joke(state: State):
    """第一次LLM调用生成初始笑话"""

    msg = llm.invoke(f"写一个关于{state['topic']}的简短笑话")
    return {"joke": msg.content}


def check_punchline(state: State):
    """门函数用于检查笑话是否有结尾”

    # 简单检查 - 笑话是否包含“？”或“！”
    if "?" in state["joke"] or "!" in state["joke"]:
        return "Fail"
    return "Pass"


def improve_joke(state: State):
    """第二次LLM调用改进笑话”

    msg = llm.invoke(f"通过添加文字游戏使这个笑话变得更有趣：{state['joke']}")
    return {"improved_joke": msg.content}


def polish_joke(state: State):
    """第三次LLM调用进行最终润色”

    msg = llm.invoke(f"在这个笑话中添加一个令人惊讶的转折：{state['improved_joke']}")
    return {"final_joke": msg.content}


# 构建工作流
workflow = StateGraph(State)

# 添加节点
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)

# 添加边连接节点
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
    "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)

# 编译
chain = workflow.compile()

# 显示工作流
display(Image(chain.get_graph().draw_mermaid_png()))

# 调用
state = chain.invoke({"topic": "cats"})
print("初始笑话：")
print(state["joke"])
print("\n--- --- ---\n")
if "improved_joke" in state:
    print("改进后的笑话：")
    print(state["improved_joke"])
    print("\n--- --- ---\n")

    print("最终笑话：")
    print(state["final_joke"])
else:
    print("笑话未通过质量门 - 没有检测到结尾！")

LangSmith跟踪

https://smith.langchain.com/public/a0281fca-3a71-46de-beee-791468607b75/r

资源：

LangChain学院

查看我们的提示链课程这里。

from langgraph.func import entrypoint, task


# 任务
@task
def generate_joke(topic: str):
    """第一次LLM调用生成初始笑话”
    msg = llm.invoke(f"写一个关于{topic}的简短笑话")
    return msg.content


def check_punchline(joke: str):
    """门函数用于检查笑话是否有结尾”
    # 简单检查 - 笑话是否包含“？”或“！”
    if "?" in joke or "!" in joke:
        return "Fail"

    return "Pass"


@task
def improve_joke(joke: str):
    """第二次LLM调用改进笑话”
    msg = llm.invoke(f"通过添加文字游戏使这个笑话变得更有趣：{joke}")
    return msg.content


@task
def polish_joke(joke: str):
    """第三次LLM调用进行最终润色”
    msg = llm.invoke(f"在这个笑话中添加一个令人惊讶的转折：{joke}")
    return msg.content


@entrypoint()
def prompt_chaining_workflow(topic: str):
    original_joke = generate_joke(topic).result()
    if check_punchline(original_joke) == "Pass":
        return original_joke

    improved_joke = improve_joke(original_joke).result()
    return polish_joke(improved_joke).result()

# 调用
for step in prompt_chaining_workflow.stream("cats", stream_mode="updates"):
    print(step)
    print("\n")

LangSmith跟踪

https://smith.langchain.com/public/332fa4fc-b6ca-416e-baa3-161625e69163/r

并行化

通过并行化，大型语言模型可以同时处理一个任务：

大型语言模型有时可以同时处理一个任务，并程序化地聚合其输出。这种工作流称为并行化，主要表现为两种关键变体：分段：将任务分解为独立的子任务并行运行；投票：多次运行同一任务以获得多样化的输出。

何时使用此工作流：当分割后的子任务可以并行化以提高速度，或者需要多个视角或尝试来获得更高置信度的结果时，使用并行化是有效的。对于具有多种考虑因素的复杂任务，大型语言模型通常在每个考虑因素由单独的大型语言模型调用处理时表现更好，这样可以专注于每个特定方面。

图API函数式API

# 图状态
class State(TypedDict):
    topic: str
    joke: str
    story: str
    poem: str
    combined_output: str


# 节点
def call_llm_1(state: State):
    """第一次大型语言模型调用生成初始笑话"""

    msg = llm.invoke(f"写一个关于{state['topic']}的主题笑话")
    return {"joke": msg.content}


def call_llm_2(state: State):
    """第二次大型语言模型调用生成故事"""

    msg = llm.invoke(f"写一个关于{state['topic']}的故事")
    return {"story": msg.content}


def call_llm_3(state: State):
    """第三次大型语言模型调用生成诗歌"""

    msg = llm.invoke(f"写一首关于{state['topic']}的诗")
    return {"poem": msg.content}


def aggregator(state: State):
    """将笑话和故事合并为单个输出"""

    combined = f"这是一个关于{state['topic']}的故事、笑话和诗歌！\n\n"
    combined += f"故事：\n{state['story']}\n\n"
    combined += f"笑话：\n{state['joke']}\n\n"
    combined += f"诗歌：\n{state['poem']}"
    return {"combined_output": combined}


# 构建工作流
parallel_builder = StateGraph(State)

# 添加节点
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)

# 添加边以连接节点
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
parallel_workflow = parallel_builder.compile()

# 显示工作流
display(Image(parallel_workflow.get_graph().draw_mermaid_png()))

# 调用
state = parallel_workflow.invoke({"topic": "猫"})
print(state["combined_output"])

LangSmith跟踪

https://smith.langchain.com/public/3be2e53c-ca94-40dd-934f-82ff87fac277/r

资源：

文档

查看我们关于并行化的文档这里。

LangChain学院

查看我们的并行化课程这里。

@task
def call_llm_1(topic: str):
    """第一次大型语言模型调用生成初始笑话"""
    msg = llm.invoke(f"写一个关于{topic}的笑话")
    return msg.content


@task
def call_llm_2(topic: str):
    """第二次大型语言模型调用生成故事"""
    msg = llm.invoke(f"写一个关于{topic}的故事")
    return msg.content


@task
def call_llm_3(topic):
    """第三次大型语言模型调用生成诗歌"""
    msg = llm.invoke(f"写一首关于{topic}的诗")
    return msg.content


@task
def aggregator(topic, joke, story, poem):
    """将笑话和故事合并为单个输出"""

    combined = f"这是一个关于{topic}的故事、笑话和诗歌！\n\n"
    combined += f"故事：\n{story}\n\n"
    combined += f"笑话：\n{joke}\n\n"
    combined += f"诗歌：\n{poem}"
    return combined


# 构建工作流
@entrypoint()
def parallel_workflow(topic: str):
    joke_fut = call_llm_1(topic)
    story_fut = call_llm_2(topic)
    poem_fut = call_llm_3(topic)
    return aggregator(
        topic, joke_fut.result(), story_fut.result(), poem_fut.result()
    ).result()

# 调用
for step in parallel_workflow.stream("猫", stream_mode="更新"):
    print(step)
    print("\n")

LangSmith跟踪

https://smith.langchain.com/public/623d033f-e814-41e9-80b1-75e6abb67801/r

路由

路由对输入进行分类并将其导向后续任务。正如Anthropic博客中关于构建有效的代理程序所述：

路由对输入进行分类，并将其导向专门的后续任务。这种工作流程允许分离关注点，并构建更专业的提示。如果没有这种工作流程，优化一种类型的输入可能会损害其他输入的表现。

使用此工作流的时机：当处理复杂任务时，其中存在不同的类别，这些类别更适合单独处理，并且分类可以准确地完成（无论是通过LLM还是传统的分类模型/算法）。

图API功能API

from typing_extensions import Literal
from langchain_core.messages import HumanMessage, SystemMessage


# 结构化输出的模式用于路由逻辑
class Route(BaseModel):
    step: Literal["poem", "story", "joke"] = Field(
        None, description="路由过程中的下一步"
    )


# 用结构化输出模式增强LLM
router = llm.with_structured_output(Route)


# 状态
class State(TypedDict):
    input: str
    decision: str
    output: str


# 节点
def llm_call_1(state: State):
    """写故事"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_2(state: State):
    """写笑话"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_3(state: State):
    """写诗"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_router(state: State):
    """将输入路由到适当的节点"""

    # 运行增强的LLM以作为路由逻辑
    decision = router.invoke(
        [
            SystemMessage(
                content="根据用户的请求将输入路由到故事、笑话或诗歌。"
            ),
            HumanMessage(content=state["input"]),
        ]
    )

    return {"decision": decision.step}


# 条件边函数以路由到适当的节点
def route_decision(state: State):
    # 返回下一个要访问的节点名称
    if state["decision"] == "story":
        return "llm_call_1"
    elif state["decision"] == "joke":
        return "llm_call_2"
    elif state["decision"] == "poem":
        return "llm_call_3"


# 构建工作流
router_builder = StateGraph(State)

# 添加节点
router_builder.add_node("llm_call_1", llm_call_1)
router_builder.add_node("llm_call_2", llm_call_2)
router_builder.add_node("llm_call_3", llm_call_3)
router_builder.add_node("llm_call_router", llm_call_router)

# 添加边以连接节点
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
    "llm_call_router",
    route_decision,
    {  # route_decision返回的名称 : 下一个要访问的节点名称
        "llm_call_1": "llm_call_1",
        "llm_call_2": "llm_call_2",
        "llm_call_3": "llm_call_3",
    },
)
router_builder.add_edge("llm_call_1", END)
router_builder.add_edge("llm_call_2", END)
router_builder.add_edge("llm_call_3", END)

# 编译工作流
router_workflow = router_builder.compile()

# 显示工作流
display(Image(router_workflow.get_graph().draw_mermaid_png()))

# 调用
state = router_workflow.invoke({"input": "给我讲个关于猫的笑话"})
print(state["output"])

LangSmith跟踪

https://smith.langchain.com/public/c4580b74-fe91-47e4-96fe-7fac598d509c/r

资源

LangChain学院

查看我们的路由课程这里。

示例

这里是RAG工作流，它会路由问题。查看我们的视频这里。

from typing_extensions import Literal
from pydantic import BaseModel
from langchain_core.messages import HumanMessage, SystemMessage


# 结构化输出的模式用于路由逻辑
class Route(BaseModel):
    step: Literal["poem", "story", "joke"] = Field(
        None, description="路由过程中的下一步"
    )


# 用结构化输出模式增强LLM
router = llm.with_structured_output(Route)


@task
def llm_call_1(input_: str):
    """写故事"""
    result = llm.invoke(input_)
    return result.content


@task
def llm_call_2(input_: str):
    """写笑话"""
    result = llm.invoke(input_)
    return result.content


@task
def llm_call_3(input_: str):
    """写诗"""
    result = llm.invoke(input_)
    return result.content


def llm_call_router(input_: str):
    """将输入路由到适当的节点"""
    # 运行增强的LLM以作为路由逻辑
    decision = router.invoke(
        [
            SystemMessage(
                content="根据用户的请求将输入路由到故事、笑话或诗歌。"
            ),
            HumanMessage(content=input_),
        ]
    )
    return decision.step


# 创建工作流
@entrypoint()
def router_workflow(input_: str):
    next_step = llm_call_router(input_)
    if next_step == "story":
        llm_call = llm_call_1
    elif next_step == "joke":
        llm_call = llm_call_2
    elif next_step == "poem":
        llm_call = llm_call_3

    return llm_call(input_).result()

# 调用
for step in router_workflow.stream("给我讲个关于猫的笑话", stream_mode="updates"):
    print(step)
    print("\n")

LangSmith跟踪

https://smith.langchain.com/public/5e2eb979-82dd-402c-b1a0-a8cceaf2a28a/r

调度器-工作者

通过调度器-工作者模式，一个调度器会将任务分解，并将每个子任务分配给不同的工作者。正如Anthropic博客中关于“构建有效的代理”所提到的：

在调度器-工作者的工作流程中，中央LLM动态地分解任务，将其委托给工作者LLM，并综合它们的结果。

何时使用此工作流：此工作流非常适合那些无法预测所需子任务的复杂任务（例如，在编码时，可能需要更改的文件数量及其每种文件的更改性质取决于具体任务）。尽管在拓扑结构上相似，但与并行化的主要区别在于其灵活性——子任务不是预先定义的，而是由调度器根据特定输入确定的。

图API功能API

from typing import Annotated, List
import operator


# 结构化输出的模式，用于规划
class Section(BaseModel):
    name: str = Field(
        description="报告该部分的名称。",
    )
    description: str = Field(
        description="该部分涵盖的主要主题和概念的简要概述。",
    )


class Sections(BaseModel):
    sections: List[Section] = Field(
        description="报告的部分。",
    )


# 使用结构化输出模式增强LLM
planner = llm.with_structured_output(Sections)

在LangGraph中创建工作者

因为调度器-工作者的工作流很常见，所以LangGraph 提供了Send API来支持这种模式。它允许你动态地创建工作者节点并向每个节点发送特定的输入。每个工作者都有自己的状态，并且所有工作者的输出都会写入一个可被调度器图访问的*共享状态键*。这使得调度器能够访问所有工作者的输出，并将其综合为最终输出。如下面所示，我们遍历一个部分列表并将每个部分发送到一个工作者节点。更多文档见这里和这里。

from langgraph.constants import Send


# 图状态
class State(TypedDict):
    topic: str  # 报告主题
    sections: list[Section]  # 报告部分列表
    completed_sections: Annotated[
        list, operator.add
    ]  # 所有工作者并行写入此键
    final_report: str  # 最终报告


# 工作者状态
class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, operator.add]


# 节点
def orchestrator(state: State):
    """生成报告计划的调度器"""

    # 生成查询
    report_sections = planner.invoke(
        [
            SystemMessage(content="生成报告计划。"),
            HumanMessage(content=f"这里是报告主题：{state['topic']}"),
        ]
    )

    return {"sections": report_sections.sections}


def llm_call(state: WorkerState):
    """工作者编写报告的一部分"""

    # 生成部分
    section = llm.invoke(
        [
            SystemMessage(
                content="按照提供的名称和描述编写报告部分。每个部分不包括前言。使用Markdown格式。"
            ),
            HumanMessage(
                content=f"这里是部分名称：{state['section'].name}和描述：{state['section'].description}"
            ),
        ]
    )

    # 将更新后的部分写入已完成部分
    return {"completed_sections": [section.content]}


def synthesizer(state: State):
    """从部分合成完整报告"""

    # 完成部分列表
    completed_sections = state["completed_sections"]

    # 格式化完成部分以字符串形式使用为最终部分的上下文
    completed_report_sections = "\n\n---\n\n".join(completed_sections)

    return {"final_report": completed_report_sections}


# 条件边函数，为每个部分创建llm_call工作者
def assign_workers(state: State):
    """为计划中的每个部分分配一个工作者"""

    # 通过Send() API并行启动部分编写
    return [Send("llm_call", {"section": s}) for s in state["sections"]]


# 构建工作流
orchestrator_worker_builder = StateGraph(State)

# 添加节点
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)

# 添加边连接节点
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges(
    "orchestrator", assign_workers, ["llm_call"]
)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)

# 编译工作流
orchestrator_worker = orchestrator_worker_builder.compile()

# 显示工作流
display(Image(orchestrator_worker.get_graph().draw_mermaid_png()))

# 调用
state = orchestrator_worker.invoke({"topic": "创建有关LLM扩展法则的报告"})

from IPython.display import Markdown
Markdown(state["final_report"])

LangSmith跟踪

https://smith.langchain.com/public/78cbcfc3-38bf-471d-b62a-b299b144237d/r

资源：

LangChain学院

查看我们的调度器-工作者课程这里。

示例

这里是一个项目，它使用调度器-工作者模式来进行报告规划和编写。查看我们的视频这里。

from typing import List


# 结构化输出的模式，用于规划
class Section(BaseModel):
    name: str = Field(
        description="报告该部分的名称。",
    )
    description: str = Field(
        description="该部分涵盖的主要主题和概念的简要概述。",
    )


class Sections(BaseModel):
    sections: List[Section] = Field(
        description="报告的部分。",
    )


# 使用结构化输出模式增强LLM
planner = llm.with_structured_output(Sections)


@task
def orchestrator(topic: str):
    """生成报告计划的调度器"""
    # 生成查询
    report_sections = planner.invoke(
        [
            SystemMessage(content="生成报告计划。"),
            HumanMessage(content=f"这里是报告主题：{topic}"),
        ]
    )

    return report_sections.sections


@task
def llm_call(section: Section):
    """工作者编写报告的一部分"""

    # 生成部分
    result = llm.invoke(
        [
            SystemMessage(content="编写报告部分。"),
            HumanMessage(
                content=f"这里是部分名称：{section.name}和描述：{section.description}"
            ),
        ]
    )

    # 将更新后的部分写入已完成部分
    return result.content


@task
def synthesizer(completed_sections: list[str]):
    """从部分合成完整报告"""
    final_report = "\n\n---\n\n".join(completed_sections)
    return final_report


@entrypoint()
def orchestrator_worker(topic: str):
    sections = orchestrator(topic).result()
    section_futures = [llm_call(section) for section in sections]
    final_report = synthesizer(
        [section_fut.result() for section_fut in section_futures]
    ).result()
    return final_report

# 调用
report = orchestrator_worker.invoke("创建有关LLM扩展法则的报告")
from IPython.display import Markdown
Markdown(report)

LangSmith跟踪

https://smith.langchain.com/public/75a636d0-6179-4a12-9836-e0aa571e87c5/r

评估器-优化器

在评估器-优化器工作流中，一个大语言模型（LLM）调用生成响应，另一个则提供评价和反馈，形成一个循环：

在评估器-优化器工作流中，一个大语言模型（LLM）调用生成响应，另一个则提供评价和反馈，形成一个循环。

何时使用此工作流：当有明确的评价标准，并且迭代改进可以带来可衡量的价值时，此工作流特别有效。两个良好的匹配标志是：首先，当人类表达他们的反馈时，大语言模型（LLM）的响应可以明显改善；其次，大语言模型（LLM）能够提供此类反馈。这类似于人类作家在制作精美的文档时可能经历的迭代写作过程。

图API函数API

# 图状态
class State(TypedDict):
    joke: str
    topic: str
    feedback: str
    funny_or_not: str


# 结构化输出的模式，用于评价
class Feedback(BaseModel):
    grade: Literal["funny", "not funny"] = Field(
        description="决定这个笑话是否有趣。",
    )
    feedback: str = Field(
        description="如果笑话不有趣，请提供如何改进它的反馈。",
    )


# 使用结构化输出模式增强大语言模型（LLM）
evaluator = llm.with_structured_output(Feedback)


# 节点
def llm_call_generator(state: State):
    """大语言模型生成笑话"""

    if state.get("feedback"):
        msg = llm.invoke(
            f"写一个关于{state['topic']}的笑话，但考虑到以下反馈：{state['feedback']}"
        )
    else:
        msg = llm.invoke(f"写一个关于{state['topic']}的笑话")
    return {"joke": msg.content}


def llm_call_evaluator(state: State):
    """大语言模型评价笑话"""

    grade = evaluator.invoke(f"评价这个笑话：{state['joke']}")
    return {"funny_or_not": grade.grade, "feedback": grade.feedback}


# 根据评估器提供的反馈路由回笑话生成器或结束
def route_joke(state: State):
    """根据评估器提供的反馈路由回笑话生成器或结束"""

    if state["funny_or_not"] == "funny":
        return "Accepted"
    elif state["funny_or_not"] == "not funny":
        return "Rejected + Feedback"


# 构建工作流
optimizer_builder = StateGraph(State)

# 添加节点
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)

# 添加边以连接节点
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
optimizer_builder.add_conditional_edges(
    "llm_call_evaluator",
    route_joke,
    {  # route_joke 返回的名字：下一个节点的名字
        "Accepted": END,
        "Rejected + Feedback": "llm_call_generator",
    },
)

# 编译工作流
optimizer_workflow = optimizer_builder.compile()

# 显示工作流
display(Image(optimizer_workflow.get_graph().draw_mermaid_png()))

# 调用
state = optimizer_workflow.invoke({"topic": "Cats"})
print(state["joke"])

LangSmith跟踪

https://smith.langchain.com/public/86ab3e60-2000-4bff-b988-9b89a3269789/r

资源

示例

这里 是一个使用评估器-优化器来改进报告的助手。观看我们的视频这里。

这里 是一个RAG工作流，它对答案进行评分，判断是否有幻觉或错误。观看我们的视频这里。

# 结构化输出的模式，用于评价
class Feedback(BaseModel):
    grade: Literal["funny", "not funny"] = Field(
        description="决定这个笑话是否有趣。",
    )
    feedback: str = Field(
        description="如果笑话不有趣，请提供如何改进它的反馈。",
    )


# 使用结构化输出模式增强大语言模型（LLM）
evaluator = llm.with_structured_output(Feedback)


# 节点
@task
def llm_call_generator(topic: str, feedback: Feedback):
    """大语言模型生成笑话"""
    if feedback:
        msg = llm.invoke(
            f"写一个关于{topic}的笑话，但考虑到以下反馈：{feedback}"
        )
    else:
        msg = llm.invoke(f"写一个关于{topic}的笑话")
    return msg.content


@task
def llm_call_evaluator(joke: str):
    """大语言模型评价笑话"""
    feedback = evaluator.invoke(f"评价这个笑话：{joke}")
    return feedback


@entrypoint()
def optimizer_workflow(topic: str):
    feedback = None
    while True:
        joke = llm_call_generator(topic, feedback).result()
        feedback = llm_call_evaluator(joke).result()
        if feedback.grade == "funny":
            break

    return joke

# 调用
for step in optimizer_workflow.stream("Cats", stream_mode="updates"):
    print(step)
    print("\n")

LangSmith跟踪

https://smith.langchain.com/public/f66830be-4339-4a6b-8a93-389ce5ae27b4/r

代理

代理通常被实现为一个在循环中根据环境反馈执行操作（通过工具调用）的大型语言模型（LLM）。正如Anthropic博客文章《构建有效的代理》中所述：

代理可以处理复杂的任务，但其实现往往很简单。它们通常是基于环境反馈在循环中使用工具的LLM。因此，设计工具集及其文档时必须清晰且周到。

使用代理的时机：当遇到难以或不可能预测所需步骤数量的开放性问题，并且无法硬编码固定路径时，可以使用代理。LLM可能会运行多个回合，您必须对其决策有一定的信任度。代理的自主性使其成为在受信任环境中扩展任务的理想选择。

API Reference: tool

from langchain_core.tools import tool


# 定义工具
@tool
def multiply(a: int, b: int) -> int:
    """乘以a和b。

    参数：
        a: 第一个整数
        b: 第二个整数
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """加a和b。

    参数：
        a: 第一个整数
        b: 第二个整数
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """除以a和b。

    参数：
        a: 第一个整数
        b: 第二个整数
    """
    return a / b


# 用工具增强LLM
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)

图API功能API

from langgraph.graph import MessagesState
from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage


# 节点
def llm_call(state: MessagesState):
    """LLM决定是否调用工具"""

    return {
        "messages": [
            llm_with_tools.invoke(
                [
                    SystemMessage(
                        content="您是一个乐于助人的助手，负责对一组输入执行算术运算。"
                    )
                ]
                + state["messages"]
            )
        ]
    }


def tool_node(state: dict):
    """执行工具调用"""

    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}


# 条件边函数，根据LLM是否进行了工具调用来路由到工具节点或结束
def should_continue(state: MessagesState) -> Literal["environment", END]:
    """根据LLM是否进行了工具调用来决定是否继续循环或停止"""

    messages = state["messages"]
    last_message = messages[-1]
    # 如果LLM进行了工具调用，则执行操作
    if last_message.tool_calls:
        return "Action"
    # 否则，我们停止（回复用户）
    return END


# 构建工作流
agent_builder = StateGraph(MessagesState)

# 添加节点
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("environment", tool_node)

# 添加连接节点的边
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    {
        # should_continue 返回的名称 : 下一个要访问的节点名称
        "Action": "environment",
        END: END,
    },
)
agent_builder.add_edge("environment", "llm_call")

# 编译代理
agent = agent_builder.compile()

# 显示代理
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# 调用
messages = [HumanMessage(content="将3和4相加。")]
messages = agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()

LangSmith跟踪

https://smith.langchain.com/public/051f0391-6761-4f8c-a53b-22231b016690/r

资源：

LangChain学院

查看我们的代理课程这里。

示例

这里是一个项目，它使用工具调用代理来创建/存储长期记忆。

from langgraph.graph import add_messages
from langchain_core.messages import (
    SystemMessage,
    HumanMessage,
    BaseMessage,
    ToolCall,
)


@task
def call_llm(messages: list[BaseMessage]):
    """LLM决定是否调用工具"""
    return llm_with_tools.invoke(
        [
            SystemMessage(
                content="您是一个乐于助人的助手，负责对一组输入执行算术运算。"
            )
        ]
        + messages
    )


@task
def call_tool(tool_call: ToolCall):
    """执行工具调用"""
    tool = tools_by_name[tool_call["name"]]
    return tool.invoke(tool_call)


@entrypoint()
def agent(messages: list[BaseMessage]):
    llm_response = call_llm(messages).result()

    while True:
        if not llm_response.tool_calls:
            break

        # 执行工具
        tool_result_futures = [
            call_tool(tool_call) for tool_call in llm_response.tool_calls
        ]
        tool_results = [fut.result() for fut in tool_result_futures]
        messages = add_messages(messages, [llm_response, *tool_results])
        llm_response = call_llm(messages).result()

    messages = add_messages(messages, llm_response)
    return messages

# 调用
messages = [HumanMessage(content="将3和4相加。")]
for chunk in agent.stream(messages, stream_mode="updates"):
    print(chunk)
    print("\n")

LangSmith跟踪

https://smith.langchain.com/public/42ae8bf9-3935-4504-a081-8ddbcbfc8b2e/r

预构建

LangGraph还提供了一种用于创建上述代理的**预构建方法**（使用create_react_agent函数）：

https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/

API Reference: create_react_agent

from langgraph.prebuilt import create_react_agent

# 传入：
# (1) 增强了工具的LLM
# (2) 工具列表（用于创建工具节点）
pre_built_agent = create_react_agent(llm, tools=tools)

# 显示代理
display(Image(pre_built_agent.get_graph().draw_mermaid_png()))

# 调用
messages = [HumanMessage(content="将3和4相加。")]
messages = pre_built_agent.invoke({"messages": messages})
for m in messages["messages"]:
    m.pretty_print()

LangSmith跟踪

https://smith.langchain.com/public/abab6a44-29f6-4b97-8164-af77413e494d/r

LangGraph提供的功能

通过在LangGraph中构建上述各项，我们获得以下几点：

持久性：人机协作

LangGraph持久层支持中断和操作批准（例如，人机协作）。参见LangChain学院模块3。

持久性：记忆

LangGraph持久层支持对话（短期）记忆和长期记忆。参见LangChain学院模块2 和模块5：

流式传输

LangGraph提供了多种方式来流式传输工作流程/代理输出或中间状态。参见LangChain学院模块3。

部署

LangGraph提供了一个简单的部署入口，支持可观察性和评估。参见LangChain学院模块6。

Comments