《AI Agent智能体开发实践+玩转FastGPT 像搭积木一样构建智能体 LLM大语言模型AI Agent开发 智能体性能优化调试部署实施方法书籍 AIAgent智能体开发实践 无规格》【摘要 书评 试读】- 京东图书

LangGraph是LangChain生态系统中的多智能体框架，它允许创建多个智能体组成的网络，这些智能体可以相互协作完成复杂任务。LangGraph提供ReAct、Plan-and-Execute、Reflection & Critique、RAG以及多智能体协作（如代理切换、群体智能）等多种模式。LangGraph定位为LangChain的扩展库，用于构建有状态的、多智能体协作系统。LangGraph的关键特性包括：

  • 基于图的拓扑结构定义工作流。
  • 支持循环和条件分支。
  • 内置持久化状态管理。
  • 兼容LangChain原有工具链。

8.6.4  多智能体框架LangGraph中的图

LangGraph中的图（Graph）可定义工作流，通过添加节点（智能体）和边（工作流转移）来构建，例如定义Supervisor、Researcher、Emailer等节点，并添加边以协调任务执行。在LangGraph中，“图”表示智能体网络的结构，由节点（智能体/工具）和边（通信路径）组成。图的结构可以是：

  • 线性图：任务按顺序执行。
  • 树状图：任务分解为子任务。
  • 有向无环图（DAG）：支持复杂的依赖关系。
  • 循环图：支持迭代或反馈机制。

【示例8.17】LangGraph基础图创建。

from langgraph.graph import StateGraph, END



# 定义状态结构（可选，但推荐）

from typing import TypedDict



class State(TypedDict):

    start: bool

    processed: bool



# 创建图

workflow = StateGraph(State)



# 定义节点函数

def start_node(state):

    print("开始执行")

    return {"start": True}



def process_node(state):

    print("处理中...")

    return {"processed": True}



# 添加节点

workflow.add_node("start", start_node)

workflow.add_node("process", process_node)



# 设置入口

workflow.set_entry_point("start")



# 添加边

workflow.add_edge("start", "process")

workflow.add_edge("process", END)      # 使用END表示结束



# 编译

app = workflow.compile()



# 执行

result = app.invoke({})

print("执行结果:", result)

示例代码执行结果如下：

开始执行

处理中...

执行结果: {'start': True, 'processed': True}

8.6.5  多智能体框架LangGraph中的状态

状态在LangGraph中表示智能体执行过程中的中间结果。每个节点（智能体）可以：

  • 读取当前状态。
  • 修改状态并传递给下一个节点。

状态可以包含：

  • 中间计算结果。
  • 上下文信息。
  • 历史决策记录。

状态管理是LangGraph的核心功能之一，确保数据在整个智能体网络中正确流动。状态管理机制示例如下：

from typing import TypedDict

class AgentState(TypedDict):

    input: str

    history: list[str]

    intermediate_steps: list[tuple]

8.6.6  多智能体框架图中的节点

节点表示智能体（如research_agent、chart_agent），每个节点有特定功能（如研究、生成图表），通过函数定义其行为，并与其他节点协作。节点是LangGraph中的基本计算单元，可以是：

  • 智能体节点：包含完整的智能体逻辑。
  • 工具节点：封装特定功能的工具。
  • 决策节点：基于条件选择路径。
  • 聚合节点：合并多个输入。

节点通过定义明确的接口与其他节点通信，接收输入状态并产生输出状态。

节点类型包括函数式节点和工具调用节点，示例如下：

# 函数式节点定义

def agent_node(state):

    response = agent.invoke(state)

    return {"response": response}

# 工具调用节点

def tool_node(state):

    tool_output = tools[state["selected_tool"]](state["input"])

    return {"output": tool_output}

【示例8.18】多节点工作流。

from langgraph.graph import StateGraph



# 定义状态类型，这在新版本中是必需的

from typing import TypedDict



class State(TypedDict):

    value: int



workflow = StateGraph(State)



def start(state: State) -> State:

    print("开始工作流")

    return {"value": 0}



def increment(state: State) -> State:

    new_value = state["value"] + 1

    print(f"增加值到 {new_value}")

    return {"value": new_value}



def end(state: State) -> State:

    print(f"最终值: {state['value']}")

    return state



workflow.add_node("start", start)

workflow.add_node("increment", increment)

workflow.add_node("end", end)



workflow.set_entry_point("start")

workflow.add_edge("start", "increment")

workflow.add_edge("increment", "end")

workflow.set_finish_point("end")



app = workflow.compile()

app.invoke({"value": 0})  # 初始状态需要符合State类型定义

输出：

开始工作流

增加值到 1

最终值: 1

8.6.7  多智能体框架LangGraph中的边

边表示智能体之间的任务转移，可通过条件边（如根据状态决定下一个节点）或直接边实现，例如Researcher完成后返回Supervisor。

（1）边表示节点之间的连接关系，它定义了：数据流动的方向、触发条件、可选的数据转换、权重（在有多个路径时使用）。边的配置决定了智能体网络如何响应不同的输入和状态变化。

（2）边控制逻辑：

from langgraph.graph import ConditionalEdge

def should_continue(state):

    if state["needs_more_info"]:

        return "human_help"

    return "end"



workflow.add_conditional_edges(

    "decision_node",

    should_continue,

    {"human_help": "human_input_node", "end": END}

)

（3）边类型：

  • 固定边：无条件转移。
  • 条件边：基于谓词函数路由。
  • 动态边：运行时计算目标节点。

【示例8.19】条件边示例。

from typing import TypedDict

from langgraph.graph import StateGraph, START, END



# 定义状态结构

class State(TypedDict):

    value: int



def start(state: State) -> State:

    print("启动节点")

    return {"value": 5}



def process_a(state: State) -> State:

    print("执行路径A")

    return state



def process_b(state: State) -> State:

    print("执行路径B")

    return state



def decide_next(state: State) -> str:

    if state["value"] > 3:

        return "process_a"

    return "process_b"



# 创建 StateGraph 实例

workflow = StateGraph(State)



# 添加节点

workflow.add_node("start", start)

workflow.add_node("process_a", process_a)

workflow.add_node("process_b", process_b)



# 设置入口

workflow.set_entry_point("start")



# 添加条件边（注意：现在直接返回目标节点名）

workflow.add_conditional_edges(

    "start",

    decide_next,

    {  # 可选：显式映射，但如果你的 decide_next 直接返回节点名，可以省略这个字典

        "process_a": "process_a",

        "process_b": "process_b"

    }

)



# 添加普通边

workflow.add_edge("process_a", "process_b")

workflow.add_edge("process_b", END)  # 设置结束点



# 编译并运行

app = workflow.compile()

app.invoke({})

执行代码，输出如下：

执行路径A

执行路径B

8.6.8  多智能体框架LangChain中的工具

LangChain提供工具（如Google搜索、Python函数等），可通过继承BaseTool类或@tool装饰器定义自定义工具，智能体根据工具描述调用相应的方法。在LangChain中，工具是智能体可以调用的外部   功能。

（1）常见的工具类型包括：

  • API工具：调用外部API，如天气API、股票API、定位API等。
  • Python函数工具：封装Python函数。
  • LLM工具：使用语言模型完成特定任务。
  • 搜索工具：如Google Serp API、Wolfram Alpha等。

（2）工具通过Tool类封装，需要定义：

  • 名称（name）。
  • 功能描述（description）。
  • 执行函数（func）。

（3）工具生态系统示例：

from langchain.tools import tool

@tool

def semantic_search(query: str) -> str:

    """向量数据库检索工具"""

    return vectorstore.similarity_search(query)

# 工具注册

tools = [semantic_search]

（4）高级功能：

  • 工具组合：通过Toolkit封装工具集。
  • 权限控制：工具访问权限管理。
  • IO校验：基于Pydantic的输入输出验证。

【示例8.20】自定义工具实现（LangChain+Qwen）。

#安装依赖包:pip install langchain langchain-community langchainhub dashscope

from langchain.tools import tool

from langchain.agents import AgentExecutor, create_react_agent

from langchain import hub

from langchain_community.chat_models import ChatTongyi



# 👇 核心计算函数，不被 @tool 装饰，避免递归调用工具对象

def _factorial(n: int) -> int:

    """内部阶乘计算函数，使用循环避免递归栈和工具调用问题"""

    if n < 0:

        raise ValueError("阶乘不支持负数")

    result = 1

    for i in range(1, n + 1):

        result *= i

    return result



# 👇 工具层函数，负责接收输入、类型转换、调用核心函数

@tool

def calculate_factorial(n: int) -> int:

    """计算给定数字的阶乘"""

    # 确保输入是整数（兼容字符串输入）

    if isinstance(n, str):

        n = int(n.strip())

    return _factorial(n)



# 工具列表

tools = [calculate_factorial]



# 初始化 Qwen 模型（请替换为你自己的 DashScope API Key）

llm = ChatTongyi(

    model="qwen-max",        # 也可用 qwen-turbo、qwen-plus

    temperature=0,

    dashscope_api_key=os.getenv("DASHSCOPE_API_KEY") 

    # 推荐：在终端或系统环境变量中设置 DASHSCOPE_API_KEY

)



# 拉取 ReAct 提示词模板

prompt = hub.pull("hwchase17/react")



# 创建 Agent

agent = create_react_agent(llm, tools, prompt)



# 创建执行器

executor = AgentExecutor(agent=agent, tools=tools, verbose=True)



# 执行任务

if __name__ == "__main__":

    response = executor.invoke({"input": "计算5的阶乘"})

    print("\n" + "="*50)

    print("最终结果:", response["output"])

输出：

> Entering new AgentExecutor chain...

我需要使用calculate_factorial函数来计算5的阶乘。

Action: calculate_factorial

Action Input: 5120我现在知道了5的阶乘的结果。

Final Answer: 5的阶乘是120。

> Finished chain.

==================================================

最终结果: 5的阶乘是120。

需要本书全部示例代码的小伙伴们，可以加我微信号 VXAAA999888，写明“要Agent代码”。

# 人工智能
Logo
龙虾开发者社区

小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用

更多推荐

AI 龙虾 | 对学习工作的影响和未来前瞻

AI 龙虾（OpenClaw）是能自主操控设备、完成全流程任务的 AI 智能体，核心是 “动口即执行”，正从工具升级为数字生产力单元，深刻重塑学习与工作模式。

avatar 龙虾开发者社区
cover

烁智 AI 云入门｜OpenClaw接入飞书：手把手教你搭建企业机器人

avatar 龙虾开发者社区

Kubernetes集群的搭建与DevOps实践（下）- 部署实践篇

(谨慎操作，相当于AI直接操作你的电脑)(可选)安装后PowerShell（终端管理员）开启文件权限openclaw config set tools.profile "coding" # 启用文件操作（read/write/edit）+ 执行命令?权限管理，配置应用权限飞书机器人 | OpenClaw 中文社区 - 开源免费 AI 助手 | WhatsApp/Telegram/微信自动化?飞书开

avatar 龙虾开发者社区