《AI Agent智能体开发实践+玩转FastGPT 像搭积木一样构建智能体 LLM大语言模型AI Agent开发 智能体性能优化调试部署实施方法书籍 AIAgent智能体开发实践 无规格》【摘要 书评 试读】- 京东图书
LangGraph是LangChain生态系统中的多智能体框架，它允许创建多个智能体组成的网络，这些智能体可以相互协作完成复杂任务。LangGraph提供ReAct、Plan-and-Execute、Reflection & Critique、RAG以及多智能体协作（如代理切换、群体智能）等多种模式。LangGraph定位为LangChain的扩展库，用于构建有状态的、多智能体协作系统。LangGraph的关键特性包括：
- 基于图的拓扑结构定义工作流。
- 支持循环和条件分支。
- 内置持久化状态管理。
- 兼容LangChain原有工具链。
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}状态在LangGraph中表示智能体执行过程中的中间结果。每个节点（智能体）可以：
- 读取当前状态。
- 修改状态并传递给下一个节点。
状态可以包含：
- 中间计算结果。
- 上下文信息。
- 历史决策记录。
状态管理是LangGraph的核心功能之一，确保数据在整个智能体网络中正确流动。状态管理机制示例如下：
from typing import TypedDict class AgentState(TypedDict): input: str history: list[str] intermediate_steps: list[tuple]节点表示智能体（如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边表示智能体之间的任务转移，可通过条件边（如根据状态决定下一个节点）或直接边实现，例如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 执行路径BLangChain提供工具（如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代码”。
