> ## Documentation Index
> Fetch the complete documentation index at: https://langchain.idochub.dev/llms.txt
> Use this file to discover all available pages before exploring further.

# 工作流与智能体

本指南回顾了常见的工作流和智能体模式。

* 工作流具有预定义的代码路径，设计为按特定顺序运行。
* 智能体是动态的，可自行定义其处理流程和工具使用方式。

[智能体工作流](/oss/images/agent_workflow.png)

LangGraph 在构建智能体和工作流时提供多项优势，包括[持久化](/oss/python/langgraph/persistence)、[流式处理](/oss/python/langgraph/streaming)、调试支持以及[部署](/oss/python/langgraph/deploy)功能。

## 设置

要构建工作流或智能体，您可以使用[任何支持结构化输出和工具调用的聊天模型](/oss/python/integrations/chat)。以下示例使用 Anthropic：

1. 安装依赖项：

```bash theme={null}
pip install langchain_core langchain-anthropic langgraph
```

2. 初始化 LLM：

```python theme={null}
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 及您为其添加的各种增强功能。[工具调用](/oss/python/langchain/tools)、[结构化输出](/oss/python/langchain/structured-output)和[短期记忆](/oss/python/langchain/short-term-memory)是几种可根据您的需求定制 LLM 的选项。

[LLM 增强功能](/oss/images/augmented_llm.png)

```python theme={null}

# 结构化输出的模式
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 调用处理前一个调用的输出。它通常用于执行可分解为更小、可验证步骤的明确定义任务。一些示例包括：

* 将文档翻译成不同语言
* 验证生成内容的一致性

[提示链式调用](/oss/images/prompt_chain.png)

<CodeGroup>
  ```python 图形 API theme={null}
  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 "通过"
      return "失败"


  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, {"失败": "improve_joke", "通过": 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": "猫"})
  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("笑话未通过质量检查 - 未检测到笑点！")
  ```

  ```python 函数式 API theme={null}
  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 "失败"

      return "通过"


  @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) == "通过":
          return original_joke

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

  # 调用
  for step in prompt_chaining_workflow.stream("猫", stream_mode="updates"):
      print(step)
      print("\n")
  ```
</CodeGroup>

## 并行化

在并行化中，LLM 同时处理一个任务。这可以通过同时运行多个独立的子任务，或多次运行同一任务以检查不同输出来实现。并行化通常用于：

* 将子任务拆分并并行运行，以提高速度
* 多次运行任务以检查不同输出，以提高置信度

一些示例包括：

* 运行一个子任务处理文档中的关键词，同时运行另一个子任务检查格式错误
* 多次运行一个基于不同标准（如引用数量、使用来源数量和来源质量）对文档准确性进行评分的任务

[parallelization.png](/oss/images/parallelization.png)

<CodeGroup>
  ```python 图形 API theme={null}
  # 图状态
  class State(TypedDict):
      topic: str
      joke: str
      story: str
      poem: str
      combined_output: str


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

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


  def call_llm_2(state: State):
      """第二次 LLM 调用，生成故事"""

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


  def call_llm_3(state: State):
      """第三次 LLM 调用，生成诗歌"""

      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"])
  ```

  ```python 函数式 API theme={null}
  @task
  def call_llm_1(topic: str):
      """第一次 LLM 调用，生成初始笑话"""
      msg = llm.invoke(f"写一个关于 {topic} 的笑话")
      return msg.content


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


  @task
  def call_llm_3(topic):
      """第三次 LLM 调用，生成诗歌"""
      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="updates"):
      print(step)
      print("\n")
  ```
</CodeGroup>

## 路由

路由工作流处理输入，然后将其引导至特定于上下文的任务。这允许您为复杂任务定义专门的流程。例如，一个用于回答产品相关问题的工作流可能首先处理问题类型，然后将请求路由到定价、退款、退货等特定流程。

[routing.png](/oss/images/routing.png)

<CodeGroup>
  ```python 图形 API theme={null}
  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"])
  ```

  ```python 函数式 API theme={null}
  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")
  ```
</CodeGroup>

## 协调器-工作者

在协调器-工作者配置中，协调器：

* 将任务分解为子任务
* 将子任务委派给工作者
* 将工作者的输出综合为最终结果

[worker.png](/oss/images/worker.png)

协调器-工作者工作流提供了更大的灵活性，通常用于子任务无法像[并行化](#parallelization)那样预先定义的情况。这在编写代码或需要跨多个文件更新内容的工作流中很常见。例如，一个需要在未知数量的文档中更新多个Python库安装说明的工作流可能会使用这种模式。

<CodeGroup>
  ```python 图形API theme={null}
  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)
  ```

  ```python 函数式API theme={null}
  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)
  ```
</CodeGroup>

### 在LangGraph中创建工作者

协调器-工作者工作流很常见，LangGraph对其有内置支持。`Send` API允许你动态创建工作者节点并向它们发送特定输入。每个工作者都有自己的状态，所有工作者的输出都会写入一个共享状态键，该键可供协调器图访问。这使协调器能够访问所有工作者的输出，并允许其将它们综合为最终输出。下面的示例遍历一个部分列表，并使用`Send` API将每个部分发送给一个工作者。

```python theme={null}
from langgraph.types 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"])
```

## 评估器-优化器

在评估器-优化器工作流中，一个LLM调用创建响应，另一个LLM调用评估该响应。如果评估器或[人在回路](/oss/python/langgraph/add-human-in-the-loop)确定响应需要改进，则提供反馈并重新创建响应。此循环持续进行，直到生成可接受的响应。

当任务有特定的成功标准，但需要迭代才能满足该标准时，通常会使用评估器-优化器工作流。例如，在两种语言之间翻译文本时，并不总是存在完美的匹配。可能需要几次迭代才能生成在两种语言中含义相同的翻译。

[evaluator\_optimizer.png](/oss/images/evaluator_optimizer.png)

<CodeGroup>
  ```python 图形API theme={null}
  # 图状态
  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):
      """LLM生成笑话"""

      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):
      """LLM评估笑话"""

      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": "猫"})
  print(state["joke"])
  ```

  ```python 函数式API theme={null}
  # 用于评估中结构化输出的模式
  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):
      """LLM生成笑话"""
      if feedback:
          msg = llm.invoke(
              f"写一个关于{topic}的笑话，但请考虑以下反馈：{feedback}"
          )
      else:
          msg = llm.invoke(f"写一个关于{topic}的笑话")
      return msg.content


  @task
  def llm_call_evaluator(joke: str):
      """LLM评估笑话"""
      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("猫", stream_mode="updates"):
      print(step)
      print("\n")
  ```
</CodeGroup>

## 智能体

智能体通常被实现为使用[工具](/oss/python/langchain/tools)执行操作的LLM。它们在持续的反馈循环中运行，用于问题和解决方案不可预测的情况。智能体比工作流具有更多的自主性，可以决定使用哪些工具以及如何解决问题。你仍然可以定义可用的工具集和智能体行为的指导方针。

[agent.png](/oss/images/agent.png)

<Note>
  要开始使用智能体，请参阅[快速入门](/oss/python/langchain/quickstart)或阅读有关LangChain中[它们如何工作](/oss/python/langchain/agents)的更多信息。
</Note>

```python 使用工具 theme={null}
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)
```

<CodeGroup>
  ```python 图形API theme={null}
  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["tool_node", END]:
      """根据LLM是否进行工具调用决定是否继续循环或停止"""

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


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

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

  # 添加边以连接节点
  agent_builder.add_edge(START, "llm_call")
  agent_builder.add_conditional_edges(
      "llm_call",
      should_continue,
      ["tool_node", END]
  )
  agent_builder.add_edge("tool_node", "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()
  ```

  ```python 函数式API theme={null}
  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")
  ```
</CodeGroup>