一、协调器-工作者
协调器:将任务分解成若干个子任务,然后将若干个子任务分配给任意个执行者,最后根据这些执行者的输出汇总结果
协调器可以具备分解和汇总的能力
Q:那么如何知道工作者到底要存在多少个?
A:根据Send API的“Send”来判断,一个节点并非只是代表一个工作者,Send相当于为这个节点创造实例也就是工作者。
二、评估器-优化器
评估器-优化器工作流通常用于任务有特定成功标准但需要迭代才能达成的情况。类似于审计功能
一、协调器-工作者
协调器:将任务分解成若干个子任务,然后将若干个子任务分配给任意个执行者,最后根据这些执行者的输出汇总结果
协调器可以具备分解和汇总的能力
Q:那么如何知道工作者到底要存在多少个?
A:根据Send API的“Send”来判断,一个节点并非只是代表一个工作者,Send相当于为这个节点创造实例也就是工作者。
二、评估器-优化器
评估器-优化器工作流通常用于任务有特定成功标准但需要迭代才能达成的情况。类似于审计功能
工作流是预定的代码路径,智能体能自主定义流程。构建它们需使用支持结构化输出和工具调用的聊天模型,并借助LLM的工具调用、结构化输出和短期记忆等增强能力。
提示链将任务分解为可验证步骤,每个LLM调用处理前一个输出。并行化同时处理多个子任务或多次运行同一任务以提升速度或可信度。路由根据输入内容将任务引导至特定语境处理。
1、工作流和智能体
工作流是具有预定性质的代码路径,并按照给定的顺序运行
而智能体是动态的,其能够自主定义其流程和工具的使用
2、环境的准备
要构建工作流和智能体,可以使用任何支持结构化输出和工具调用的聊天模型
参考LangGraph快速开始,主要是依赖项的安装和LLM的初始化
3、LLMs与增强
工作流和智能体系统都基于LLMs和为其添加的增强技能,比如工具调用、结构化输出、短期记忆
主要是对Harness具备的能力的选择和模型的选择
4、提示链
提示链是指每一个LLM调用处理前一个调用的输出,它通常是指用于执行定义明确、可分解为更小可验证步骤的任务。
如,输入给我查询xxx一周的天气情况,根据这一周的每日最高温度和最低温度画出折线图
LLM分析查询xxx一周天气,然后将结果和下一步的操作作为下一个节点的输入传输给下一个节点
{"city":"北京","dates":["周一"...], "highs":[25,26...], "lows":[15,16...]} (只传数据,不传操作)5、并行化
通过并行化,LLMs可以同时处理一项任务,具体表现为同时运行多个独立的子任务,或是多次运行同一任务以检查不同的输出结果来实现,因此并行化通常用于
(1)将子任务拆分并并行执行,从而提升执行速度
(2)多次运行任务来检查是否存在不同的输出,从而提高可信度
6、路由
路由工作流会对输入进行处理,并将其引导至特定语境下的任务
步骤二:明确每个步骤需要做什么
对于图中的每个节点,确定它代表什么类型的操作以及它需要什么上下文才能正常工作。
1、LLM处理:需要理解、分析、生成文本或做出推理决策
2、数据流向:需要从外部来源检索的信息
3、执行流程:需要执行外部的操作
4、人工干涉:需要人工干预的步骤
步骤三:设计状态
状态是智能体中所有节点均可访问的共享记忆
1、它是否需要在多个步骤之间持久存在?如果是,就放进状态字典中
2、它是否可以从其他数据推导出来?如果是,则在需要时计算它,而不是将其存储在 状态中
状态仅包含原始数据——不包含提示模板、格式化字符串或任何指令。分类输出直接来自 LLM,并以单个字典的形式存储。(根据实用主义来看,这个是全图共享但是使用还是相关节点或边)
步骤 4:构建节点
将每个步骤实现为一个函数。LangGraph 中的一个节点就是一个 Python 函数,它接收当前状态并返回更新后的状态。
| 错误类型 | 谁来修复 | 策略 | 适用场景 |
|---|---|---|---|
| 临时性错误(网络问题、速率限制) | 系统(自动) | 重试策略 | 通常重试即可解决的临时故障 |
| LLM 可恢复的错误(工具调用失败、解析问题) | LLM | 将错误信息存入 State 并循环返回 | LLM 能够看到错误并调整自己的处理方式 |
| 用户可修复的错误(信息缺失、指令不清晰) | 人类 | 使用 interrupt() 暂停执行 | 需要用户提供额外信息才能继续 |
| 重试用尽后的可恢复失败 | 开发者 | error_handler | 在重试全部失败后,执行补偿/恢复分支逻辑 |
| 意外错误 | 开发者 | 冒泡 | 未知问题,需要调试介入 |
1、临时性错误:可通过添加重试策略,自动重试网络问题和速率限制问题。
2、LLM可恢复错误:将错误存储在状态中并循环返回,以便 LLM 可以查看哪里出了问题并重试
3、用户可介入错误:必要时暂停并从用户处收集信息
4、重试用尽错误:运行恢复功能,更新状态并路由到补偿分支
5、意外错误:冒泡调试
步骤五:将他们组合在一起
构建graph
需注意节点颗粒度权衡
1、考虑隔离外部服务
2、中间节点可见
3、不同的故障模型
4、可用性、测试性,要注意重用性
1、首先确定自己设备的python版本
python -V
2、创建虚拟环境(此要求python版本≥3.11)
#在适合的地方创建venv文件夹
python -m venv venv
#激活虚拟环境(此目录为例)
E:\langgraph_project\venv\Scripts\activate
#激活成功后将会为(venv) E:\langgraph_project>
# 升级 pip
python -m pip install –upgrade pip
# 安装 langgraph-cli
pip install -U “langgraph-cli[inmem]”
pip install langchain-openai langgraph
3、验证安装是否成功
langgraph –help
4、创建一个 LangGraph 应用(例项目名为first_agent,路径为E:\langgraph_project\first_agent)
langgraph new first_agent –template new-langgraph-project-python
5、安装依赖
cd first_agent
pip install -e .
6、项目关联至pycharm
以下操作在pycharm终端进行
7、配置.env
cp .env.example .env
# .env
LANGSMITH_API_KEY=lsv2…
8、依赖安装
uv sync
8、启动服务
langgraph dev
如果启动不起来,可尝试一下chrome的隐私模式
1、pycharm新建项目,纯python项目->解释器类型uv,位置自定义,然后点击创建(python版本不低于3.10)

2、点击终端,依次运行以下指令
(.venv) PS E:\xxx\xx> uv add langgraph
(.venv) PS E:\xxx\xx> uv add langchain
3、复制快速入门中源码
# Step 1: Define tools and model
from langchain.tools import tool
from langchain.chat_models import init_chat_model
model = init_chat_model(
“claude-sonnet-4-6”,
temperature=0
)
# Define tools
@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
# Augment the LLM with tools
tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)
# Step 2: Define state
from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], operator.add]
llm_calls: int
# Step 3: Define model node
from langchain.messages import SystemMessage
def llm_call(state: MessagesState):
“””LLM decides whether to call a tool or not”””
return {
“messages”: [
model_with_tools.invoke(
[
SystemMessage(
content=”You are a helpful assistant tasked with performing arithmetic on a set of inputs.”
)
]
+ state[“messages”]
)
],
“llm_calls”: state.get(‘llm_calls’, 0) + 1
}
# Step 4: Define tool node
from langchain.messages import ToolMessage
def tool_node(state: MessagesState):
“””Performs the tool call”””
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}
# Step 5: Define logic to determine whether to end
from typing import Literal
from langgraph.graph import StateGraph, START, END
# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: MessagesState) -> Literal[“tool_node”, END]:
“””Decide if we should continue the loop or stop based upon whether the LLM made a tool call”””
messages = state[“messages”]
last_message = messages[-1]
# If the LLM makes a tool call, then perform an action
if last_message.tool_calls:
return “tool_node”
# Otherwise, we stop (reply to the user)
return END
# Step 6: Build agent
# Build workflow
agent_builder = StateGraph(MessagesState)
# Add nodes
agent_builder.add_node(“llm_call”, llm_call)
agent_builder.add_node(“tool_node”, tool_node)
# Add edges to connect nodes
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”)
# Compile the agent
agent = agent_builder.compile()
from IPython.display import Image, display
# Show the agent
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))
# Invoke
from langchain.messages import HumanMessage
messages = [HumanMessage(content=”Add 3 and 4.”)]
messages = agent.invoke({“messages”: messages})
for m in messages[“messages”]:
m.pretty_print()
以deepseek V4连接为例
根据编写之日看deepseek说:init_chat_model 对 DeepSeek 的“原生”支持不如 ChatOpenAI 稳定,因此强烈建议直接使用 ChatOpenAI
所以上述代码咱们需要变更一处内容
将
model = init_chat_model(
“claude-sonnet-4-6”,
temperature=0
)
修改为以下内容
from langchain_openai import ChatOpenAI
model = ChatOpenAI(
model=”deepseek-chat”, #模型根据官方接口文档进行修正
api_key=”sk-…”, # 替换为你的实际 key
base_url=”https://api.deepseek.com”,
temperature=0
)
之后只需要再安装langchain_openai和IPython至项目中即可通过运行main.py,就可以得到以下输出内容:
================================ Human Message =================================
Add 3 and 4.
================================== Ai Message ==================================
Tool Calls:
add (call_00_xxxxxxxxxxx)
Call ID: call_00_xxxxxxxxxxx
Args:
a: 3
b: 4
================================= Tool Message =================================
7
================================== Ai Message ==================================
The sum of 3 and 4 is **7**.
langgraph 是一个底层编排框架和运行时,用于构建、管理和部署长时间运行的有状态代理
具备的功能:持久执行、流式传输、人机交互
核心优势
持久性:能够在故障发生后还能长时间运行,并且可以在中断后可以以之前状态继续运行。
人机交互:通过在任意时刻检查和修改智能体状态,来融入人工监督。
全面记忆:构建有状态智能体,兼具用于持续推理的短期工作记忆,以及跨会话的长期记忆。
使用 LangSmith 调试:借助可视化工具深度透视复杂智能体行为,这些工具可追踪执行路径、捕获状态变迁,并提供详细的运行时指标。
准生产部署:利用可扩展的基础设施,可靠地部署复杂智能体系统,该设施专为应对有状态、长时运行工作流的特殊挑战而构建。
应用程序结构(至少是这样的)
my-app/
├── my_agent # all project code lies within here
│ ├── utils # utilities for your graph
│ │ ├── __init__.py
│ │ ├── tools.py # tools for your graph
│ │ ├── nodes.py # node functions for your graph
│ │ └── state.py # state definition of your graph
│ ├── __init__.py
│ └── agent.py # code for constructing your graph
├── .env # environment variables
├── requirements.txt # package dependencies
└── langgraph.json # configuration file for LangGraph
配置文件
langgraph的配置文件为langgraph.json里面指定了部署langgraph应用程序所需的依赖变量和其他设置
人工智能体的介绍会很快略过,后面有机会再补漏吧(笑)以下AI Agent会简称为Agent
Q:Agent是什么?
A:人工智能体,一种能够感知环境进行自主理解,进行决策和执行动作的智能体
Q:Agent是有什么组成
A: Agent = Model + Harness
Q:那什么是Model?
A:Mode就是大模型
Q:那什么是Harness
A:那就是马具,用来骑马的,🐴(你说的都对)
A:Harness其实是指模型外层的工程化运行框架。它通过执行环境、工具接口、上下文控制、任务编排、可观测性、评估反馈和治理机制,将模型调用组织成可执行、可控制、可追踪的任务流程。详细的请看这个网页👉(https://mp.weixin.qq.com/s?__biz=Mzg4MDE3OTA5NA==&mid=2247605175&idx=1&sn=b7209ecde854b9bae309d9e9d8ca8846&chksm=cedd890177d87942f3403505290b5d0869955b3bed7013e10904b964baf9ffb4cd478d4f5ab2&scene=0&xtrack=1)
如果你看过25年那张AI Agent图的话,你会感觉确实有这个意思
不过后续再细聊吧
这是一次略带莽撞的尝试。起初,这台服务器专为部署 Openclaw 而生,寄望于乘上人工智能的浪潮。然而现实清冷,Openclaw 的实际用量远不及预期,服务器也随之陷入闲置。
既然硬件已经就位,与其任其空转,不如让它换一种方式继续运转——搭建一个记录学习轨迹的平台。哪怕最终结不出什么硕果,至少它见证过探索的痕迹,也承载过一段不虚掷的光阴。