多 Agent 协同方案实践：基于 LangGraph 搭建能源领域智能调度工作流

版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

摘要

随着电力市场、虚拟电厂、储能调度业务复杂度提升，单一大模型难以完成数据采集、电价分析、调度策略生成、风险校验等链式任务。本文基于LangGraph构建多Agent协同架构，拆分数据爬虫Agent、RAG知识库Agent、调度决策Agent、校验Agent，实现能源场景自动化工作流。文中完整讲解状态图原理、环境部署、可运行Python工程代码，融合RAG、Function Calling、非结构化文档解析技术，为电力/新能源行业多智能体落地提供工程参考。
关键词：LangGraph、多Agent、大模型应用、能源调度、RAG、智能体工作流

[此处插入图片：能源领域多Agent协同整体架构图，包含4个智能体、状态流转、向量数据库、外部数据源交互链路]

一、行业背景与技术选型

1.1 能源调度场景痛点

传统新能源、虚拟电厂调度系统依赖固定规则脚本，存在明显短板：

普通顺序调用LangChain Chain无法实现条件分支、循环重试、任务中断，而AutoGen、CrewAI侧重于对话式多智能体，缺少可视化状态管控能力。

1.2 LangGraph核心优势

LangGraph 在LangChain基础上引入**状态机（State Graph）**思想，核心特性：

• 全局统一State对象，所有Agent共享上下文；
• 支持有向图定义任务节点、条件分支、循环流程；
• 兼容RAG、Function Calling、工具调用，无缝对接向量数据库；
• 支持流程断点、持久化，适合长链路能源业务流程。

1.3 整体架构设计

本文搭建4个职能分离的Agent：

流程流转：采集Agent → 检索Agent → 决策Agent → 条件判断：校验通过输出结果 / 校验失败，重新生成调度方案。

[此处插入图片：LangGraph状态流转流程图，节点：采集Agent、RAG Agent、决策Agent、校验Agent，标注分支回流逻辑]

二、开发环境搭建

2.1 依赖库清单

新建requirements.txt

langgraph==0.2.27
langchain==0.3.10
langchain-openai==0.2.11
langchain-community==0.3.10
transformers==4.45.2
torch==2.4.1
chromadb==0.5.11
pypdf==4.3.1
beautifulsoup4==4.12.3
python-dotenv==1.0.2

执行安装命令

pip install -r requirements.txt

说明：代码同时兼容通义千问Qwen、DeepSeek等开源模型，仅需替换LLM初始化配置；向量数据库选用Chroma用于本地测试，生产环境可替换PostgreSQL pgvector。

2.2 环境变量配置

创建.env文件，填入模型密钥与参数

LLM_BASE_URL=https://api.deepseek.com/v1
LLM_API_KEY=your_api_key
MODEL_NAME=deepseek-chat
VECTOR_DB_PATH=./energy_vector_db

三、核心代码实现

3.1 全局状态State定义

LangGraph依靠State在节点之间传递数据，我们自定义能源调度专属状态，包含采集数据、知识库上下文、调度方案、校验结果。

from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage

# 定义全局状态结构体
class EnergyScheduleState(TypedDict):
messages: Annotated[Sequence[BaseMessage], operator.add]
# 爬虫获取的电价、新能源出力原始数据
market_data: str
# RAG检索得到的电力政策文档片段
policy_context: str
# 生成的储能调度方案
schedule_plan: str
# 合规校验结果
check_result: str
# 是否通过校验标识
is_approved: bool

3.2 初始化LLM与向量库RAG工具

import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.prompts import PromptTemplate

# 加载环境变量
load_dotenv()

# 初始化大模型
llm = ChatOpenAI(
base_url=os.getenv("LLM_BASE_URL"),
api_key=os.getenv("LLM_API_KEY"),
model=os.getenv("MODEL_NAME"),
temperature=0.1
)

# 初始化Embedding与向量数据库
embedding_model = HuggingFaceEmbeddings(
model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vector_store = Chroma(
persist_directory=os.getenv("VECTOR_DB_PATH"),
embedding_function=embedding_model
)
retriever = vector_store.as_retriever(search_kwargs={"k": 3})

# RAG检索Prompt模板
rag_prompt = PromptTemplate.from_template("""
你是电力行业政策专家，根据检索到的文档回答调度相关约束条件。
【检索文档】：{context}
用户问题：{query}
只输出和储能、电力交易调度相关的约束规则，不要额外扩充。
""")

3.3 定义四大Agent节点函数

3.3.1 Agent1：数据采集智能体（模拟爬虫）

def data_collect_agent(state: EnergyScheduleState) –> EnergyScheduleState:
"""
数据采集Agent：模拟爬虫获取日前电价、光伏出力数据
工程中可替换为真实网页爬虫、API接口、OCR文档解析
"""
print("=====【数据采集Agent】开始执行 =====")
# 模拟爬虫采集结果，真实场景使用BeautifulSoup解析电力交易平台网页
market_info = """
日前市场电价：
08:00-12:00 电价 0.62元/kWh
14:00-18:00 电价 0.55元/kWh
20:00-23:00 电价 0.71元/kWh
光伏预测出力：午间10MW，晚间出力下降至2MW
储能容量：5MWh，最大充放电功率2MW
"""
return {
"market_data": market_info
}

3.3.2 Agent2：知识库检索RAG智能体

def knowledge_retrieve_agent(state: EnergyScheduleState) –> EnergyScheduleState:
"""
RAG知识库Agent：检索向量库中电力交易、储能调度政策PDF文档
"""
print("=====【知识库检索Agent】开始执行 =====")
query = "储能参与电力市场充放电约束、电价套利相关政策要求"
docs = retriever.get_relevant_documents(query)
context_text = "\\n".join([doc.page_content for doc in docs])

chain = rag_prompt | llm
response = chain.invoke({"context": context_text, "query": query})
return {
"policy_context": response.content
}

3.3.3 Agent3：调度决策智能体

from langchain_core.messages import HumanMessage

def schedule_decision_agent(state: EnergyScheduleState) –> EnergyScheduleState:
"""
调度方案生成Agent：结合市场数据+政策约束生成储能调度策略
"""
print("=====【调度决策Agent】开始执行 =====")
prompt = f"""
【实时市场数据】
{state['market_data']}
【电力政策约束】
{state['policy_context']}
任务：生成一份储能充放电调度方案，明确各时段充放电功率、套利逻辑。
"""
resp = llm.invoke([HumanMessage(content=prompt)])
return {
"schedule_plan": resp.content
}

3.3.4 Agent4：合规校验智能体

def compliance_check_agent(state: EnergyScheduleState) –> EnergyScheduleState:
"""
合规校验Agent：判断调度方案是否满足政策约束
"""
print("=====【合规校验Agent】开始执行 =====")
check_prompt = f"""
政策约束：{state['policy_context']}
待校验调度方案：{state['schedule_plan']}
任务：
1. 判断方案是否符合储能调度相关政策；
2. 输出结论：【通过】或【不通过】，同时写明理由；
3. 如果不通过，指出需要修改的地方。
"""
resp = llm.invoke([HumanMessage(content=check_prompt)])
result_text = resp.content
approved = "【通过】" in result_text
return {
"check_result": result_text,
"is_approved": approved
}

3.4 定义分支判断函数与LangGraph工作流

from langgraph.graph import StateGraph, END

def route_check_result(state: EnergyScheduleState):
"""
条件分支路由：校验通过直接结束；不通过回流到决策Agent重新生成方案
"""
if state["is_approved"]:
return END
else:
return "schedule_decision_agent"

# 构建状态图
workflow = StateGraph(EnergyScheduleState)

# 添加所有节点
workflow.add_node("data_collect_agent", data_collect_agent)
workflow.add_node("knowledge_retrieve_agent", knowledge_retrieve_agent)
workflow.add_node("schedule_decision_agent", schedule_decision_agent)
workflow.add_node("compliance_check_agent", compliance_check_agent)

# 设置流程入口
workflow.set_entry_point("data_collect_agent")

# 线性链路
workflow.add_edge("data_collect_agent", "knowledge_retrieve_agent")
workflow.add_edge("knowledge_retrieve_agent", "schedule_decision_agent")
workflow.add_edge("schedule_decision_agent", "compliance_check_agent")

# 条件分支
workflow.add_conditional_edges(
"compliance_check_agent",
route_check_result,
{
"schedule_decision_agent": "schedule_decision_agent",
END: END
}
)

# 编译图
app = workflow.compile()

# 启动执行
if __name__ == "__main__":
init_state = {
"messages": [],
"market_data": "",
"policy_context": "",
"schedule_plan": "",
"check_result": "",
"is_approved": False
}
final_state = app.invoke(init_state)
print("\\n====================最终调度方案====================")
print(final_state["schedule_plan"])
print("\\n====================校验报告====================")
print(final_state["check_result"])

[此处插入图片：代码运行输出示例截图，展示Agent依次执行日志、最终调度方案与校验结果]

四、工程落地踩坑经验总结

4.1 State数据传递常见问题

4.2 RAG与Agent协同优化

能源场景存在大量PDF、扫描件，需要OCR预处理非结构化文档：

• 扫描版PDF先调用OCR工具提取文本，再切片向量化入库；
• 检索不要只依赖相似度，增加元数据过滤（政策发布时间、适用区域），避免过时规则干扰调度决策。

4.3 生产环境优化建议

4.4 业务扩展方向

当前案例为4个基础Agent，后续可以扩展：

• 新增电价预测Agent，接入时序预测模型；
• 新增可视化Agent，自动生成调度图表；
• 对接储能设备API，实现方案下发闭环。

五、结语

本文基于LangGraph实现了能源调度场景多Agent协同工作流，将复杂业务拆解为职责单一的智能体，利用状态机实现条件分支、循环重试，解决单一大模型无法胜任长链路业务的痛点。整套架构兼容RAG、网页爬虫、非结构化文档解析，可直接迁移至虚拟电厂、电力交易、新能源储能等业务系统。

多Agent目前仍处于工程落地阶段，下一阶段重点方向是多智能体通信协议优化、自主分工博弈、长任务记忆管理。欢迎各位从事能源AI、大模型应用开发的开发者在评论区交流：

本文完整代码已上传，如有需要可私信获取向量库文档预处理脚本、PDF-OCR解析扩展模块。