> ## Documentation Index > Fetch the complete documentation index at: https://nvd-54.mintlify.site/llms.txt > Use this file to discover all available pages before exploring further. # 工作流与智能体 本指南介绍常见的工作流和智能体模式。 * 工作流具有预定义的代码路径，设计为按特定顺序执行。 * 智能体是动态的，能自主定义执行流程和工具使用方式。 智能体工作流 LangGraph 在构建智能体和工作流时提供了多种优势，包括[持久化](/oss/python/langgraph/persistence)、[流式输出](/oss/python/langgraph/streaming)，以及对调试和[部署](/oss/python/langgraph/deploy)的支持。 使用 [LangSmith](https://smith.langchain.com?utm_source=docs\&utm_medium=cta\&utm_campaign=langsmith-signup\&utm_content=oss-langgraph-workflows-agents) 追踪和比较这些工作流模式。按照[追踪快速入门](/langsmith/trace-with-langgraph)了解数据如何在每个步骤中流转。 ## 设置 要构建工作流或智能体，你可以使用任何支持结构化输出和工具调用的[聊天模型](/oss/python/integrations/chat)。以下示例使用 Anthropic： 1. 安装依赖： ```bash theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} pip install langchain_core langchain-anthropic langgraph ``` 2. 初始化大语言模型(LLM)： ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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-sonnet-4-6") ``` ## 大语言模型(LLM)与增强功能 工作流和智能体系统基于大语言模型(LLM)以及你为其添加的各种增强功能。[工具调用](/oss/python/langchain/tools)、[结构化输出](/oss/python/langchain/structured-output)和[短期记忆](/oss/python/langchain/short-term-memory)是定制 LLM 以满足需求的几种选择。 LLM 增强功能 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 结构化输出的 Schema from pydantic import BaseModel, Field class SearchQuery(BaseModel): search_query: str = Field(None, description="Query that is optimized web search.") justification: str = Field( None, description="Why this query is relevant to the user's request." ) # 使用结构化输出 schema 增强 LLM structured_llm = llm.with_structured_output(SearchQuery) # 调用增强后的 LLM output = structured_llm.invoke("How does Calcium CT score relate to high cholesterol?") # 定义工具 def multiply(a: int, b: int) -> int: return a * b # 使用工具增强 LLM llm_with_tools = llm.bind_tools([multiply]) # 使用触发工具调用的输入调用 LLM msg = llm_with_tools.invoke("What is 2 times 3?") # 获取工具调用结果 msg.tool_calls ``` ## 提示链 提示链是指每次 LLM 调用处理前一次调用的输出。它通常用于执行可以分解为更小、可验证步骤的明确任务。一些示例包括： * 将文档翻译成不同语言 * 验证生成内容的一致性 提示链 ```python Graph API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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"Write a short joke about {state['topic']}") return {"joke": msg.content} def check_punchline(state: State): """门控函数检查笑话是否有点睛之笔""" # 简单检查——笑话是否包含 "?" 或 "!" if "?" in state["joke"] or "!" in state["joke"]: return "Pass" return "Fail" def improve_joke(state: State): """第二次 LLM 调用改进笑话""" msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}") return {"improved_joke": msg.content} def polish_joke(state: State): """第三次 LLM 调用进行最终润色""" msg = llm.invoke(f"Add a surprising twist to this joke: {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("最终笑话:") print(state["joke"]) ``` ```python Functional API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.func import entrypoint, task # 任务 @task def generate_joke(topic: str): """第一次 LLM 调用生成初始笑话""" msg = llm.invoke(f"Write a short joke about {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"Make this joke funnier by adding wordplay: {joke}") return msg.content @task def polish_joke(joke: str): """第三次 LLM 调用进行最终润色""" msg = llm.invoke(f"Add a surprising twist to this joke: {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") ``` ## 并行化 通过并行化，多个 LLM 同时处理一个任务。这可以是同时运行多个独立子任务，也可以是多次运行同一任务以检查不同输出。并行化常用于： * 将子任务拆分并行运行，提升速度 * 多次运行任务检查不同输出，提高置信度 一些示例包括： * 一个子任务处理文档关键词提取，另一个子任务检查格式错误 * 多次运行一个评分任务，根据不同标准（如引用数量、使用的来源数量和来源质量）为文档准确性打分 parallelization.png ```python Graph API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 图状态 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"Write a joke about {state['topic']}") return {"joke": msg.content} def call_llm_2(state: State): """第二次 LLM 调用生成故事""" msg = llm.invoke(f"Write a story about {state['topic']}") return {"story": msg.content} def call_llm_3(state: State): """第三次 LLM 调用生成诗歌""" msg = llm.invoke(f"Write a poem about {state['topic']}") return {"poem": msg.content} def aggregator(state: State): """将笑话、故事和诗歌合并为单一输出""" combined = f"Here's a story, joke, and poem about {state['topic']}!\n\n" combined += f"STORY:\n{state['story']}\n\n" combined += f"JOKE:\n{state['joke']}\n\n" combined += f"POEM:\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": "cats"}) print(state["combined_output"]) ``` ```python Functional API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} @task def call_llm_1(topic: str): """第一次 LLM 调用生成笑话""" msg = llm.invoke(f"Write a joke about {topic}") return msg.content @task def call_llm_2(topic: str): """第二次 LLM 调用生成故事""" msg = llm.invoke(f"Write a story about {topic}") return msg.content @task def call_llm_3(topic): """第三次 LLM 调用生成诗歌""" msg = llm.invoke(f"Write a poem about {topic}") return msg.content @task def aggregator(topic, joke, story, poem): """将笑话和故事合并为单一输出""" combined = f"Here's a story, joke, and poem about {topic}!\n\n" combined += f"STORY:\n{story}\n\n" combined += f"JOKE:\n{joke}\n\n" combined += f"POEM:\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("cats", stream_mode="updates"): print(step) print("\n") ``` ## 路由 路由工作流处理输入并将其定向到特定上下文的任务。这允许你为复杂任务定义专门的流程。例如，一个回答产品相关问题的工作流可能先处理问题类型，然后将请求路由到定价、退款、退货等特定流程。 routing.png ```python Graph API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from typing_extensions import Literal from langchain.messages import HumanMessage, SystemMessage # 使用结构化输出作为路由逻辑的 Schema class Route(BaseModel): step: Literal["poem", "story", "joke"] = Field( None, description="The next step in the routing process" ) # 使用结构化输出 schema 增强 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="Route the input to story, joke, or poem based on the user's request." ), 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": "Write me a joke about cats"}) print(state["output"]) ``` ```python Functional API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from typing_extensions import Literal from pydantic import BaseModel from langchain.messages import HumanMessage, SystemMessage # 使用结构化输出作为路由逻辑的 Schema class Route(BaseModel): step: Literal["poem", "story", "joke"] = Field( None, description="The next step in the routing process" ) # 使用结构化输出 schema 增强 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="Route the input to story, joke, or poem based on the user's request." ), 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("Write me a joke about cats", stream_mode="updates"): print(step) print("\n") ``` ## 编排者-工作者 在编排者-工作者配置中，编排者： * 将任务分解为子任务 * 将子任务委派给工作者 * 将工作者输出综合为最终结果 worker.png 编排者-工作者工作流提供更多灵活性，通常在子任务无法像[并行化](#parallelization)那样预定义时使用。这在需要编写代码或更新多个文件内容的工作流中很常见。例如，一个需要在未知数量的文档中更新多个 Python 库安装说明的工作流可能会使用此模式。 ```python Graph API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from typing import Annotated, List import operator # 用于规划的结构化输出 Schema class Section(BaseModel): name: str = Field( description="Name for this section of the report.", ) description: str = Field( description="Brief overview of the main topics and concepts to be covered in this section.", ) class Sections(BaseModel): sections: List[Section] = Field( description="Sections of the report.", ) # 使用结构化输出 schema 增强 LLM planner = llm.with_structured_output(Sections) ``` ```python Functional API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from typing import List # 用于规划的结构化输出 Schema class Section(BaseModel): name: str = Field( description="Name for this section of the report.", ) description: str = Field( description="Brief overview of the main topics and concepts to be covered in this section.", ) class Sections(BaseModel): sections: List[Section] = Field( description="Sections of the report.", ) # 使用结构化输出 schema 增强 LLM planner = llm.with_structured_output(Sections) @task def orchestrator(topic: str): """编排者生成报告计划""" # 生成查询 report_sections = planner.invoke( [ SystemMessage(content="Generate a plan for the report."), HumanMessage(content=f"Here is the report topic: {topic}"), ] ) return report_sections.sections @task def llm_call(section: Section): """工作者编写报告的一个章节""" # 生成章节 result = llm.invoke( [ SystemMessage(content="Write a report section."), HumanMessage( content=f"Here is the section name: {section.name} and description: {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("Create a report on LLM scaling laws") from IPython.display import Markdown Markdown(report) ``` ### 在 LangGraph 中创建工作者 编排者-工作者工作流很常见，LangGraph 对其提供了内置支持。`Send` API 允许你动态创建工作者节点并发送特定输入。每个工作者有自己的状态，所有工作者输出写入编排者图可访问的共享状态键。这使编排者能够访问所有工作者输出并将它们综合为最终输出。下面的示例遍历章节列表，使用 `Send` API 将每个章节发送给一个工作者。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} 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="Generate a plan for the report."), HumanMessage(content=f"Here is the report topic: {state['topic']}"), ] ) return {"sections": report_sections.sections} def llm_call(state: WorkerState): """工作者编写报告的一个章节""" # 生成章节 section = llm.invoke( [ SystemMessage( content="Write a report section following the provided name and description. Include no preamble for each section. Use markdown formatting." ), HumanMessage( content=f"Here is the section name: {state['section'].name} and description: {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": "Create a report on LLM scaling laws"}) from IPython.display import Markdown Markdown(state["final_report"]) ``` ## 评估者-优化者 在评估者-优化者工作流中，一个 LLM 调用创建响应，另一个评估该响应。如果评估者或[人机协作](/oss/python/langgraph/interrupts)环节确定响应需要改进，则提供反馈并重新创建响应。此循环持续进行直到生成可接受的响应。 评估者-优化者工作流通常在任务有特定成功标准但需要迭代才能满足该标准时使用。例如，在两种语言之间翻译文本时并不总能完美匹配。可能需要几次迭代才能生成在两种语言中具有相同含义的翻译。 evaluator_optimizer.png ```python Graph API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 图状态 class State(TypedDict): joke: str topic: str feedback: str funny_or_not: str # 用于评估的结构化输出 Schema class Feedback(BaseModel): grade: Literal["funny", "not funny"] = Field( description="Decide if the joke is funny or not.", ) feedback: str = Field( description="If the joke is not funny, provide feedback on how to improve it.", ) # 使用结构化输出 schema 增强 LLM evaluator = llm.with_structured_output(Feedback) # 节点 def llm_call_generator(state: State): """LLM 生成笑话""" if state.get("feedback"): msg = llm.invoke( f"Write a joke about {state['topic']} but take into account the feedback: {state['feedback']}" ) else: msg = llm.invoke(f"Write a joke about {state['topic']}") return {"joke": msg.content} def llm_call_evaluator(state: State): """LLM 评估笑话""" grade = evaluator.invoke(f"Grade the joke {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"]) ``` ```python Functional API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} # 用于评估的结构化输出 Schema class Feedback(BaseModel): grade: Literal["funny", "not funny"] = Field( description="Decide if the joke is funny or not.", ) feedback: str = Field( description="If the joke is not funny, provide feedback on how to improve it.", ) # 使用结构化输出 schema 增强 LLM evaluator = llm.with_structured_output(Feedback) # 节点 @task def llm_call_generator(topic: str, feedback: Feedback): """LLM 生成笑话""" if feedback: msg = llm.invoke( f"Write a joke about {topic} but take into account the feedback: {feedback}" ) else: msg = llm.invoke(f"Write a joke about {topic}") return msg.content @task def llm_call_evaluator(joke: str): """LLM 评估笑话""" feedback = evaluator.invoke(f"Grade the joke {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") ``` ## 智能体 智能体通常作为使用[工具](/oss/python/langchain/tools)执行操作的 LLM 来实现。它们在持续反馈循环中运行，用于问题和解决方案不可预测的场景。智能体比工作流具有更多自主性，可以自行决定使用哪些工具以及如何解决问题。你仍然可以定义可用的工具集和智能体行为准则。 agent.png 要开始使用智能体，请参阅[快速入门](/oss/python/langchain/quickstart)或阅读更多关于 LangChain 中智能体[工作原理](/oss/python/langchain/agents)的内容。 ```python 使用工具 theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.tools import tool # 定义工具 @tool def multiply(a: int, b: int) -> int: """Multiply `a` and `b`. Args: a: First int b: Second int """ return a * b @tool def add(a: int, b: int) -> int: """Adds `a` and `b`. Args: a: First int b: Second int """ return a + b @tool def divide(a: int, b: int) -> float: """Divide `a` and `b`. Args: a: First int b: Second int """ 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) ``` ```python Graph API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import MessagesState from langchain.messages import SystemMessage, HumanMessage, ToolMessage # 节点 def llm_call(state: MessagesState): """LLM 决定是否调用工具""" return { "messages": [ llm_with_tools.invoke( [ SystemMessage( content="You are a helpful assistant tasked with performing arithmetic on a set of inputs." ) ] + 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="Add 3 and 4.")] messages = agent.invoke({"messages": messages}) for m in messages["messages"]: m.pretty_print() ``` ```python Functional API theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langgraph.graph import add_messages from langchain.messages import ( SystemMessage, HumanMessage, ToolCall, ) from langchain_core.messages import BaseMessage @task def call_llm(messages: list[BaseMessage]): """LLM 决定是否调用工具""" return llm_with_tools.invoke( [ SystemMessage( content="You are a helpful assistant tasked with performing arithmetic on a set of inputs." ) ] + 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="Add 3 and 4.")] for chunk in agent.stream(messages, stream_mode="updates"): print(chunk) print("\n") ``` ### ToolNode [`ToolNode`](https://reference.langchain.com/python/langgraph/agents/#langgraph.prebuilt.tool_node.ToolNode) 是一个预构建的节点，用于在 LangGraph 工作流中执行工具。它自动处理并行工具执行、错误处理和状态注入。 当你需要对图如何执行工具进行细粒度控制时，使用 [`ToolNode`](https://reference.langchain.com/python/langgraph/agents/#langgraph.prebuilt.tool_node.ToolNode)。这是许多 LangGraph 智能体模式中工具执行的基础构建块。 ```python theme={"theme":{"light":"catppuccin-latte","dark":"catppuccin-mocha"}} from langchain.tools import tool from langgraph.prebuilt import ToolNode from langgraph.graph import MessagesState, StateGraph @tool def search(query: str) -> str: """Search for information.""" return f"Results for: {query}" @tool def calculator(expression: str) -> str: """Evaluate a math expression.""" return str(eval(expression)) builder = StateGraph(MessagesState) builder.add_node("tools", ToolNode([search, calculator])) # ... 添加其他节点和边 graph = builder.compile() ``` ***
[连接这些文档](/use-these-docs)到 Claude、VSCode 等工具，通过 MCP 获取实时回答。 [在 GitHub 上编辑此页面](https://github.com/langchain-ai/docs/edit/main/src/oss/langgraph/workflows-agents.mdx)或[提交问题](https://github.com/langchain-ai/docs/issues/new/choose)。