LangChain 中的智能体是其最强大的功能之一，它允许大语言模型根据用户输入动态决定执行哪些操作，通过调用工具（Tools ）来完成任务。智能体将LLM 作为推理引擎，使其能够自主决策和执行复杂任务。

智能体介绍

智能体（Agent ）以大语言模型（LLM ）为“大脑”，根据用户输入和当前情境，从工具箱中选择合适的工具执行操作，实现自动化决策和行动，可应用于客户服务、智能家居等场景。

人工智能代理是一种能够感知环境、自主决策并执行动作的智能程序。在LangChain 框架中，智能体可以理解为“能够基于工具和知识库自主执行任务的AI 系统”。在LangChain 框架中，AI Agent 通常由以下几个核心组件构成。

l   工具（Tools ）：Agent 可以调用的功能模块。

l   记忆（Memory ）：存储历史交互信息。

l   决策引擎：决定下一步采取什么行动。

l   执行器：实际执行选择的行动。

中的AI Agent 实现方式

在LangChain 中，Agent 基类提供了智能体的基础结构。

1. 核心组件

from langchain.agents import AgentExecutor, create _ react _ agent

from langchain import hub

# 基础实现流程

prompt = hub.pull("hwchase17/react")  # 获取预设提示模板

agent = create _ react _ agent(llm, tools, prompt)  # 绑定 LLM 与工具

agent _ executor = AgentExecutor(agent=agent, tools=tools)  # 创建可执行智能体

2. 智能体类型

l   Zero-shot React ：基于ReAct 框架的零样本推理。

l   Conversational ：专为对话优化的智能体。

l   Plan-and-execute ：先规划再执行的层次化智能体。

多智能体框架LangGraph 介绍

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

l   基于图的拓扑结构定义工作流。

l   支持循环和条件分支。

l   内置持久化状态管理。

l   兼容LangChain 原有工具链。

多智能体框架LangGraph 中的图

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

l   线性图：任务按顺序执行。

l   树状图：任务分解为子任务。

l   有向无环图（DAG ）：支持复杂的依赖关系。

l   循环图：支持迭代或反馈机制。

【示例 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}

多智能体框架LangGraph 中的状态

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

l   读取当前状态。

l   修改状态并传递给下一个节点。

状态可以包含：

l   中间计算结果。

l   上下文信息。

l   历史决策记录。

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

from typing import TypedDict

class AgentState(TypedDict):

    input: str

    history: list[str]

    intermediate _ steps: list[tuple]

多智能体框架图中的节点

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

l   智能体节点：包含完整的智能体逻辑。

l   工具节点：封装特定功能的工具。

l   决策节点：基于条件选择路径。

l   聚合节点：合并多个输入。

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

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

# 函数式节点定义

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

多智能体框架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 ）边类型：

l   固定边：无条件转移。

l   条件边：基于谓词函数路由。

l   动态边：运行时计算目标节点。

【示例 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

多智能体框架LangChain 中的工具

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

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

l   API 工具：调用外部API ，如天气API 、股票API 、定位API 等。

l   Python 函数工具：封装Python 函数。

l   LLM 工具：使用语言模型完成特定任务。

l   搜索工具：如Google Serp API 、Wolfram Alpha 等。

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

l   名称（name ）。

l   功能描述（description ）。

l   执行函数（func ）。

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

from langchain.tools import tool

@tool

def semantic _ search(query: str) -> str:

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

    return vectorstore.similarity _ search(query)

# 工具注册

tools = [semantic _ search]

（4 ）高级功能：

l   工具组合：通过Toolkit 封装工具集。

l   权限控制：工具访问权限管理。

l   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 。

本文 节选自《AI Agent智能体开发实践》，获出版社和作者授权发布。