本文系统讲解如何基于 LangGraph 框架搭建可落地的 Agent 工作流，涵盖工作流核心逻辑拆解、Tavily 搜索引擎工具集成、状态定义与动态重规划功能实现全流程。结合 DeepSeek 大模型与 Tavily 实时搜索能力，从环境配置、代码编写到效果验证，为新手程序员提供一站式指导，帮你快速掌握复杂任务智能 Agent 的构建方法。

1 、Agent工作流

先明确核心定义：

Agent 工作流是指多个智能体（Agent）遵循预设逻辑顺序与协作规则，借助各类工具完成复杂任务的闭环流程体系。不同于单一 AI 调用，Agent 工作流具备 “自主决策” 能力，是构建智能应用的核心框架。

LangGraph（LangChain 生态核心组件）实现 Agent 工作流的核心特点：

• 智能规划：LLM 将复杂任务（如 “分析 2025 年新能源汽车销量趋势”）拆解为可执行的小步骤，明确每一步的目标与执行路径；
• 工具执行：调用外部工具（搜索引擎、数据库、API 等）完成具体任务，支持权限管控与工具调用校验；
• 反思迭代：每步执行后自动评估结果是否符合预期，若未达标则动态调整计划，直到完成目标；
• 状态留存：全程记录任务输入、执行步骤、中间结果，支持断点续跑与流程追溯（新增特性）。

简言之，Agent 工作流的核心是 “先规划、再执行、常复盘”，完成单步任务后可根据实际结果调整后续计划，而非按固定脚本执行。

2、 Tavily工具

Tavily是一个为大型语言模型（LLMs）和检索增强生成（RAG）优化的搜索引擎，旨在提供高效、快速且持久的搜索结果。本案例中我们需要申请Tavily API Key。

• Tavily 官网：https://www.tavily.com/
• Tavily API：Tavily API是一款专为AI应用设计的实时搜索引擎接口，提供高速、准确且事实性的搜索结果，支持每月1000次免费查询，并可通过Python集成到LangChain等框架中。

3、 LangGraph构建Agent工作流

3.1 安装工具包

首先我们需要安装所需的软件包。以下是安装LangGraph的步骤：

pip install --quiet -U langgraph langchain-community langchain-openai tavily-python

3.2 导入LangGraph和LangChain模块

在使用LangChain之前，需要导入LangChain x OpenAI集成包，并设置API密钥作为环境变量或直接传递给OpenAI LLM类，本案例使用的是DeepSeek模型。

首先，获取DeepSeek的API密钥(需要购买)，当然也可以使用阿里百炼模型模型创建账户并申请API来获取。然后，可以将API密钥设置为环境变量，方法如下：

import os

from langchain import hub

from langchain_deepseek import ChatDeepSeek

from langgraph.prebuilt import create_react_agent

from langchain_community.tools.tavily_search import TavilySearchResults

# 设置TAVILY和DeepSeek API

os.environ["DEEPSEEK_API_KEY"] = "替换DeepSeek API"

os.environ["TAVILY_API_KEY"] = "替换第二步申请的Tavily API"

以上是关于LangGraph和LangChain模块的导入，以及TAVILY和DeepSeek的API设置。

3.3 创建智能代理执行器

首先从LangChain Hub获取预定义的prompt模板，结合Tavily搜索工具和DeepSeek语言模型，完成REACT代理执行器。

# 1. 从langchain的hub获取prompt的模板

prompt = hub.pull("zhang1career/react-agent-executor")

prompt.pretty_print()

# 2. 创建TavilySearchResults工具，设置最大结果数为1

tools = [TavilySearchResults(max_results = 1)]

# 3. 选择llm

llm = ChatDeepSeek(

    model="deepseek-chat",

    api_key=os.getenv("DEEPSEEK_API_KEY"),

)

# 4. 创建一个REACT代理执行器，使用指定的LLM和工具，并应用从Hub中获取的prompt 

agent_executor = create_react_agent(llm, tools,prompt=prompt)

3.4 定义状态和规划步骤

定义状态和规划步骤的目的在于指导任务执行。通过PlanExecute类存储任务的输入、计划、过往步骤和响应，提供任务执行的完整状态记录。Plan类则用于描述未来要执行的计划，明确按顺序排列的步骤。此外，创建了planner_prompt提示模板，用于生成简单、逐步且独立的任务计划，确保每一步包含必要信息，最终直接得出答案。

import operator

from pydantic import BaseModel, Field

from typing import Annotated, List, Tuple, TypedDict,Literal

from langchain_core.prompts import ChatPromptTemplate

# 定义一个TYpeDict类 PlanExecute , 用于存储输入、计划、过去的步骤和响应

class PlanExecute(TypedDict):

    input: str

    plan:List[str]

    past_steps:Annotated[List[Tuple], operator.add]

    response:str

# 定义一个plan模型类， 用于描述未来要执行的计划

classPlan(BaseModel):

    """未来要执行的计划"""

    steps:List[str] = Field(

        description="需要执行的不同步骤，应该按顺序排列"

    )

# 创建一个计划生成的提示模板

planner_prompt = ChatPromptTemplate.from_messages(

    [

        (

            "system",

            """对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确答案，不要添加任何多余的步骤，最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 不要跳过步骤。"""

        ),

        ("placeholder","{messages}")

    ]

)

# 使用指定的提示词模板创建一个计划生成器，使用deepseek模型

planner = planner_prompt|ChatDeepSeek(

   model="deepseek-chat",

    api_key=os.getenv("DEEPSEEK_API_KEY"),

).with_structured_output(Plan)

3.5 重新规划步骤

现在让我们考虑创建规划步骤。 这将使用函数调用来创建计划，并实现了任务的重新规划功能。

from typing import Union

# 定义一个响应模型类，用于描述用户的响应

class Response(BaseModel):

    """用户响应"""

    response: str

# 定义一个行为模型类，用于描述要执行的行为，该类继承自BaseModel

class Act(BaseModel):

    """要执行的行为"""

    action: Union[Response,Plan] = Field(

        description="要执行的行为， 如果要回应用户，使用Response，如果需要进一步使用工具获取答案，使用Plan。"

    )

# 重新规划步骤提示词 

replanner_prompt = ChatPromptTemplate.from_template(

    """对于给定的目标，提出一个简单的逐步计划。这个计划应该包含独立的任务，如果正确执行将得出正确答案，不要添加任何多余的步骤，最后一步的结果应该是最终答案。确保每一步都有所有必要的信息 - 

    你的目标是：

    {input}

    你的原计划是：

    {plan}

    你目前已完成的步骤是：

    {past_steps}

    相应的更新你的计划，如果不需要更多的步骤并且可以返回给用户，那么就这样响应。如果需要，填写计划。只添加仍然需要完成的步骤，不要返回已完成的步骤

    """

)   

# 使用指定的提示模板创建一个重新计划生成器，使用deepseek

replanner = replanner_prompt|ChatDeepSeek(

   model="deepseek-chat",

    api_key=os.getenv("DEEPSEEK_API_KEY"),

).with_structured_output(Act)

3.6 创建工作流图

接下来，我们基于Langgraph框架的异步工作流，用于处理复杂任务的规划、执行和动态调整。它通过定义异步函数分别负责生成计划、执行步骤和重新规划任务，构建了一个完整的任务处理流程。

import asyncio

asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

# 定义一个异步函数

async def main():

    # 定义一个异步函数，用于生成计划步骤

    async def plan_step(state: PlanExecute):

        plan = awaitplanner.ainvoke({"messages": [("user", state["input"])]})

        return {"plan": plan.steps}

    # 定义一个异步函数执行步骤

    async def execute_step(state: PlanExecute):

        plan = state["plan"]

        plan_str = "\n".join(f"{i+1}.{step}"fori, stepinenumerate(plan))

        task = plan[0]

        task_formatted = f"""对于一下计划:

         {plan_str}\n\n你的任务执行第{1}步，{task}.

           """

        agent_response = awaitagent_executor.ainvoke(

            {"messages":[("user", task_formatted)]}

        )

        return {"past_steps": state["past_steps"]+[(task, agent_response["messages"][-1].content)]}

    # 定义一个异步函数，用于重新计划步骤

    async def replan_step(state: PlanExecute):

        output = awaitreplanner.ainvoke(state)

        if isinstance(output.action, Response):

            return {"response": output.action.response}

        else:

             return {"plan":output.action.steps}

    # 定义一个函数用于判断是否结束

    def should_end(state: PlanExecute)->Literal["agent","__end__"]:

        if("response"instateandstate["response"]):

            return"__end__"

        else:

            return"agent"

    from langgraph.graph import StateGraph, START

    # 创建一个状态图，初始化PlanEexecute

    workflow = StateGraph(PlanExecute)

    # 添加计划节点

    workflow.add_node("planner",plan_step)

    # 添加执行步骤节点

    workflow.add_node("agent", execute_step)

    # 重新计划节点

    workflow.add_node("replan", replan_step)

    # 从开始到计划节点的边

    workflow.add_edge(START, "planner")

    # 设置从计划到代理节点的边

    workflow.add_edge("planner", "agent")

    # 设置从代理到重新计划节点的边

    workflow.add_edge("agent","replan")

    # 添加条件边，用于判断下一步操作

    workflow.add_conditional_edges(

        "replan",

        should_end,

    )

    app = workflow.compile()

    graph_png = app.get_graph().draw_mermaid_png()

    with open("agent_workflow.png", "wb") as f:

        f.write(graph_png)

    # 设置配置，递归限制为50次

    config = {"recursion_limit":50}

    inputs = {"input":"2025年WTT乒乓球男单冠军是是谁？请用中文回答"}

    asyncfor event in app.astream(inputs, config=config):

        for k,v in event.items():

            if k!="__end__":

                print(v)

asyncio.run(main())

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套 AI 大模型突围资料包：

• ✅ 从零到一的 AI 学习路径图
• ✅ 大模型调优实战手册（附医疗/金融等大厂真实案例）
• ✅ 百度/阿里专家闭门录播课
• ✅ 大模型当下最新行业报告
• ✅ 真实大厂面试真题
• ✅ 2026 最新岗位需求图谱

所有资料 ⚡️ ，朋友们如果有需要 《AI大模型入门+进阶学习资源包》，下方扫码获取~

① 全套AI大模型应用开发视频教程

（包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点）

② 大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

③ 大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

④ AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

⑤ 大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

⑥ 大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

以上资料如何领取？

为什么大家都在学大模型？

最近科技巨头英特尔宣布裁员2万人，传统岗位不断缩减，但AI相关技术岗疯狂扩招，有3-5年经验，大厂薪资就能给到50K*20薪！

不出1年，“有AI项目经验”将成为投递简历的门槛。

风口之下，与其像“温水煮青蛙”一样坐等被行业淘汰，不如先人一步，掌握AI大模型原理+应用技术+项目实操经验，“顺风”翻盘！

这些资料真的有用吗？

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

以上全套大模型资料如何领取？