动态增加agent node
LangGraph中如何实现动态增加agent node
在 LangChain / LangGraph 里实现“队长动态加成员”核心思路
- 队长维护一个成员注册表（内存/数据库）
- 队长判断：当前任务缺对应专长 → 生成新角色+新system_prompt → 注册新成员 → 调度它干活
- 成员是运行时临时创建的子 Agent，不用提前写死所有角色
下面分：架构设计 → 动态创建流程 → 可直接跑的代码（极简版） → 提示词模板 → FAQ。
一、整体架构（适配你的需求）
1. 角色
- 队长（Supervisor）：唯一常驻；负责拆任务、匹配成员、创建新成员、汇总结果
- 常驻成员（固定池）：文字专员、工具调用专员（默认就有）
- 动态成员（临时池）：图片、音频、数据、代码…（队长按需创建，用完可保留/销毁）
2. 核心数据结构（状态）
用 LangGraph 的 State 存：
{
"task": str, # 当前总任务
"members": dict, # 成员注册表：{name: {role, system_prompt, tools, ...}}
"sub_tasks": list, # 已拆解子任务
"results": dict, # 各成员返回结果
"next": str # 下一个要执行的节点（队长/某成员/结束）
}
二、队长“动态创建成员”的完整流程（5步）
-
队长接收任务 → 分析需要哪些专长
- 例：用户发了一张图+问题 → 需要「图片分析」能力
-
查成员注册表 → 有没有对应专长的成员？
- 有 → 直接分配任务
- 无 → 进入“创建新成员”流程
-
队长生成新成员的核心配置（AI自主生成）
- 角色名：
图片视觉专员_001（带序号防重） - 身份+能力+边界：用统一模板生成
- system_prompt：按模板填充
- 工具：绑定图片相关工具（如 OCR、图像描述）
- 角色名：
-
注册新成员到成员表
state["members"]["图片视觉专员_001"] = { "role": "图片视觉专员", "system_prompt": "...", "tools": [image_tools], "status": "idle" } -
队长分配任务给新成员 → 执行 → 收结果 → 汇总
三、可直接运行的极简代码（LangGraph + create_agent）
1. 依赖
pip install langchain langgraph langchain-openai
2. 完整代码（可直接复制运行）
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent, AgentExecutor
from langchain_core.prompts import SystemPromptTemplate
from langgraph import StateGraph, START, END
from langgraph.typing import TypedDict, Annotated
import operator
# ---------------------- 1. 基础配置 ----------------------
llm = ChatOpenAI(model="gpt-5", temperature=0)
# 初始常驻成员（文字、工具）
INIT_MEMBERS = {
"文字专员": {
"role": "文本文案专员",
"system_prompt": "你是团队专职文本文案专员。擅长：文案撰写、润色、摘要、结构化。不做：工具、图片、音频、拆解任务。只听队长，只输出文本结果。",
"tools": []
},
"工具专员": {
"role": "工具调用执行员",
"system_prompt": "你是团队专职工具调用执行员。擅长：调用工具、解析参数、整理结果。不做：创作、图片、音频。参数缺失直接反馈队长。",
"tools": []
}
}
# ---------------------- 2. 状态定义 ----------------------
class TeamState(TypedDict):
task: str
members: Annotated[dict, operator.or_] # 合并成员表
sub_tasks: list
results: dict
next: str
# ---------------------- 3. 队长节点（核心：创建成员+调度） ----------------------
def captain_node(state: TeamState) -> TeamState:
task = state["task"]
members = state["members"] or INIT_MEMBERS.copy()
# ---- 步骤1：分析任务需要的专长 ----
if "图片" in task or "图" in task:
required_role = "图片视觉专员"
elif "音频" in task or "语音" in task:
required_role = "音频处理专员"
else:
required_role = "文本文案专员"
# ---- 步骤2：检查是否已有该专长成员 ----
existing = [n for n, m in members.items() if m["role"] == required_role]
if not existing:
# ---- 步骤3：动态创建新成员 ----
new_name = f"{required_role}_{len(members)+1}"
# 按模板生成system_prompt
if required_role == "图片视觉专员":
new_prompt = "你是团队专职图片视觉分析员。擅长：图片识别、图文提取、画面描述、内容总结。不做：音频、文案、工具。客观描述，不编造。"
elif required_role == "音频处理专员":
new_prompt = "你是团队专职音频语音处理员。擅长：语音转文字、摘要、对话提取、场景识别。不做：图片、文案、工具。保持原意，不改写。"
else:
new_prompt = "你是团队专职文本文案专员。擅长：文案、润色、摘要、结构化。不做：工具、图片、音频。只听队长。"
# 注册新成员
members[new_name] = {
"role": required_role,
"system_prompt": new_prompt,
"tools": [],
"status": "idle"
}
print(f"✅ 队长动态创建新成员：{new_name}")
# ---- 步骤4：分配任务给对应成员（选第一个匹配的） ----
target_member = [n for n, m in members.items() if m["role"] == required_role][0]
state["sub_tasks"] = [{"to": target_member, "task": task}]
state["next"] = target_member
state["members"] = members
return state
# ---------------------- 4. 成员节点工厂（动态生成执行节点） ----------------------
def build_member_node(name: str, config: dict):
"""根据成员配置，动态创建可执行的Agent节点"""
agent = create_agent(
llm=llm,
tools=config["tools"],
system_prompt=SystemPromptTemplate.from_template(config["system_prompt"])
)
executor = AgentExecutor(agent=agent, tools=config["tools"])
def node_func(state: TeamState) -> TeamState:
# 取当前成员的任务
my_task = [t["task"] for t in state["sub_tasks"] if t["to"] == name][0]
result = executor.invoke({"input": my_task})
state["results"][name] = result["output"]
state["next"] = "captain" # 执行完回到队长
return state
return node_func
# ---------------------- 5. 构建图（动态注册成员节点） ----------------------
def build_graph():
graph = StateGraph(TeamState)
# 初始节点：队长
graph.add_node("captain", captain_node)
graph.add_edge(START, "captain")
# 动态注册所有成员节点（初始+后续新增）
def update_members(state: TeamState):
for name, config in state["members"].items():
if name not in graph.nodes:
graph.add_node(name, build_member_node(name, config))
graph.add_edge(name, "captain") # 成员干完回队长
# 队长决定下一步：成员 or 结束
if state["next"] in state["members"]:
graph.add_edge("captain", state["next"])
else:
graph.add_edge("captain", END)
return graph.compile()
# ---------------------- 6. 测试运行 ----------------------
if __name__ == "__main__":
app = build_graph()
# 测试：需要图片能力的任务（会自动创建图片专员）
result = app.invoke({
"task": "分析这张图片的内容：有什么物体、场景、文字？",
"members": INIT_MEMBERS,
"sub_tasks": [],
"results": {},
"next": "captain"
})
print("\n📌 最终结果：")
print(result["results"])
四、动态成员 System Prompt 模板（直接套用）
模板（4段式，极简清晰）
身份：你是团队【{角色名}】
擅长：{能力1}、{能力2}、{能力3}（3-6条，具体）
不做：{排除1}、{排除2}、{排除3}（明确边界）
规则：只承接队长分配的专属子任务，只输出本领域结果，不越权、不编造。
示例（动态生成时填充）
-
图片专员：
身份：你是团队【图片视觉专员】 擅长：图片内容识别、图文信息提取、画面要素描述、图像内容总结 不做：不处理音频、不做文案创作、不调用业务工具、不承接纯文字任务 规则：只承接队长分配的图片分析任务，客观描述，不臆测画面外信息。
五、FAQ（关键问题解答）
Q1：新成员创建后，会一直存在吗？
A：默认长期保留（存在 state["members"] 里），下次同类型任务可直接复用；也可设置“用完即删”（在成员节点执行完后从字典删除），节省内存。
Q2：队长怎么避免创建重复角色？
A：用「角色名+序号」命名（如 图片视觉专员_001），并在创建前先查 members 里是否已有同 role 的成员，有就复用，无则新建。
Q3：动态成员需要独立的记忆吗？
A：不需要独立记忆。所有成员的对话历史都存在全局 state 里，成员只处理当前子任务，无状态，轻量化。
Q4：可以动态给新成员分配工具吗？
A：可以。创建时在 config["tools"] 里绑定对应工具（如图片专员绑定 OCR工具、图像描述工具），队长可根据任务动态分配工具。
Q5：如何限制成员数量，避免无限创建？
A：队长加个最大成员数阈值（如 10 个），超过后不再新建，转而提示用户“任务类型过多，无法创建更多成员”。
六、关键优势（匹配你的需求）
- 队长极轻：只做调度，prompt 短，上下文少
- 成员按需扩编：缺什么能力建什么成员，不用预定义所有角色
- 职责彻底隔离：每个成员只干自己的事，不越权
- 模板化易扩展：新增角色直接套 4 段式 prompt，零成本