在带内存与路由的LangGraph工作流中集成独立KPI专属Agent的实现指导请求

我完全懂你现在的困扰——把KPI相关的工具和独立流程拆成专属Agent，确实能让整体架构更清晰、扩展性更强，但要把它无缝整合到你现有的带内存和路由的LangGraph工作流里，光看零散的文档确实容易摸不着头绪。我来给你一步步拆解具体的实现思路和代码改造方向：

首先先把你现有的核心代码贴出来方便对照：

tools = [check_user_promotion, get_local_data, get_kpi_info, list_all_kpis]
llm_with_tools = model.bind_tools(tools)

def reasoner(state: MessagesState):
    message_history = [
        msg for msg in state["messages"][:-1]  
        if isinstance(msg, (HumanMessage, AIMessage)) 
        and not hasattr(msg, 'tool_calls') 
        and not hasattr(msg, 'tool_call_id')
    ]
    
    if len(message_history) >= 4:
        last_human_message = state["messages"][-1]
        summary_prompt = (
            "Summarize the conversation between the human and AI, "
            "focusing only on the key points of their dialogue. "
            "Ignore any tool interactions or technical details."
        )
        summary_message = model.invoke(
            message_history + [HumanMessage(content=summary_prompt)]
        )

        delete_messages = [RemoveMessage(id=m.id) for m in state["messages"]]
        human_message = HumanMessage(content=last_human_message.content)
        response = llm_with_tools.invoke([sys_msg, summary_message, human_message])
        message_updates = [summary_message, human_message, response] + delete_messages
    else:
        message_updates = llm_with_tools.invoke([sys_msg] + state["messages"])
    return {"messages": message_updates}

builder = StateGraph(MessagesState)
builder.add_node("reasoner", reasoner)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "reasoner")
builder.add_conditional_edges("reasoner", tools_condition)
builder.add_edge("tools", "reasoner")
memory = MemorySaver()
react_graph = builder.compile(checkpointer=memory)

1. 先搭建独立的KPI专属Agent工作流 #

首先我们要把KPI相关的逻辑（专属工具、人类在环流程）封装成一个独立的LangGraph Agent，它有自己的状态、节点和路由规则：

# 定义KPI Agent专属的工具（可以加入你说的更多KPI相关工具）
kpi_tools = [get_kpi_info, list_all_kpis, calculate_kpi_trend, validate_kpi_data]
llm_kpi = model.bind_tools(kpi_tools)

# 定义KPI Agent的状态（可复用MessagesState，也可扩展KPI特定字段）
class KpiAgentState(MessagesState):
    kpi_context: Optional[Dict] = None

# KPI Agent的推理节点（加入专属业务逻辑）
def kpi_reasoner(state: KpiAgentState):
    user_query = state["messages"][-1].content
    # 示例：如果用户请求需要人工审核，直接路由到人类节点
    if "审核" in user_query or "人工确认" in user_query:
        return {"messages": [AIMessage(content="请人工审核以下KPI数据：")]}
    # 否则调用KPI专属工具处理
    response = llm_kpi.invoke([sys_msg] + state["messages"])
    return {"messages": [response]}

# 人类在环节点（处理需要人工干预的KPI请求）
def kpi_human_in_loop(state: KpiAgentState):
    # 这里可实现获取人工反馈的逻辑，比如等待输入后更新状态
    human_feedback = state["messages"][-1].content
    return {"messages": [AIMessage(content=f"已收到人工反馈：{human_feedback}，继续处理KPI请求")]}

# 构建KPI Agent的工作流
kpi_builder = StateGraph(KpiAgentState)
kpi_builder.add_node("kpi_reasoner", kpi_reasoner)
kpi_builder.add_node("kpi_tools", ToolNode(kpi_tools))
kpi_builder.add_node("kpi_human", kpi_human_in_loop)

# 定义KPI Agent内部的路由条件
def kpi_route_condition(state: KpiAgentState):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, 'tool_calls'):
        return "kpi_tools"
    elif "审核" in last_msg.content or "人工确认" in last_msg.content:
        return "kpi_human"
    else:
        return END

# 连接KPI Agent的节点
kpi_builder.add_edge(START, "kpi_reasoner")
kpi_builder.add_conditional_edges("kpi_reasoner", kpi_route_condition)
kpi_builder.add_edge("kpi_tools", "kpi_reasoner")
kpi_builder.add_edge("kpi_human", "kpi_reasoner")

# 给KPI Agent加上内存
kpi_memory = MemorySaver()
kpi_agent = kpi_builder.compile(checkpointer=kpi_memory)

2. 将KPI Agent整合进主工作流 #

把刚才的KPI Agent作为一个节点加入到你原来的主工作流中，实现基于请求类型的智能路由：

# 主工作流中新增KPI Agent节点（包装一层确保状态同步）
def run_kpi_agent(state: MessagesState):
    # 把主工作流的消息传递给KPI Agent
    kpi_state = KpiAgentState(messages=state["messages"])
    # 运行KPI Agent直到流程结束
    final_kpi_state = kpi_agent.invoke(kpi_state)
    # 把KPI处理结果同步回主工作流的消息池
    return {"messages": final_kpi_state["messages"]}

builder.add_node("kpi_agent", run_kpi_agent)

# 修改主工作流的路由条件：判断用户请求是否和KPI相关
def main_route_condition(state: MessagesState):
    last_user_msg = state["messages"][-1].content.lower()
    # 可替换为LLM分类或更精准的关键词匹配
    if any(keyword in last_user_msg for keyword in ["kpi", "关键绩效指标", "绩效数据"]):
        return "kpi_agent"
    # 保留原有的通用工具路由逻辑
    last_ai_msg = [msg for msg in state["messages"] if isinstance(msg, AIMessage)][-1]
    if hasattr(last_ai_msg, 'tool_calls'):
        return "tools"
    else:
        return END

# 更新主工作流的条件路由
builder.add_conditional_edges("reasoner", main_route_condition)
# KPI Agent处理完后回到主reasoner节点，继续后续对话
builder.add_edge("kpi_agent", "reasoner")

3. 确保内存与状态的一致性 #

为了避免对话上下文断裂，需要保证主工作流和KPI Agent的状态同步：

• 复用同一个MemorySaver实例，让主工作流和KPI Agent共享内存池
• 通过包装函数run_kpi_agent实现主状态与KPI Agent状态的双向同步
• 可以在KPI Agent的结束逻辑中，把关键处理结果提炼成摘要消息，避免主工作流消息池过于臃肿

4. 测试与调整 #

完成改造后，重点测试两种核心场景：

• 非KPI请求：仍然走原来的reasoner和通用工具流程，保持原有对话逻辑不变
• KPI相关请求：自动路由到KPI专属Agent，执行它自己的工具调用、人类在环流程，处理完后无缝回到主对话流

这样既保留了你原来的工作流逻辑，又把KPI相关的复杂流程拆分到独立Agent中，扩展性和维护性都会好很多。

备注：内容来源于stack exchange，提问作者sultan aljahwary