LangGraph 多智能体实战：搭建电力政策跟踪 Agent，实现网页爬虫、OCR 与 PDF 文档自动化解析

文章前言

随着国内电力市场化改革持续推进，各省电力交易中心持续发布交易规则、政策文件、市场通知。电力从业者需要持续跟踪海量PDF公告、网页政策文本，人工整理、对比政策差异、提取核心条款成本极高。传统单一大模型应用无法完成网页采集→文档解析→文本抽取→知识入库→智能分析完整流水线。

本文基于LangGraph构建多智能体协同系统，拆分爬虫智能体、文档解析智能体、规则抽取智能体、知识库更新智能体。整套工作流自动完成网页抓取、PDF解析、图片OCR识别、政策条款结构化抽取，结合RAG向量存储实现电力政策持续跟踪。整套代码可直接工程改造，适配能源、电力交易业务场景。

技术栈：Python3.10+ | LangGraph | LangChain | Pytesseract(OCR) | PyPDF2 | BeautifulSoup | Chroma向量库 | OpenAI兼容大模型接口

一、系统整体架构设计

1.1 多智能体角色划分

我们将整个业务流程拆解为4个独立Agent，由LangGraph负责状态流转、路由调度：

1.2 工作流流转逻辑

起始节点 → 爬虫Agent → 文档下载分发 → 文档解析Agent（分支：网页/PDF/OCR图片）
→ 规则抽取Agent → 知识库写入Agent → 任务结束
出现异常自动重试，解析失败文档标记，存入异常队列。

1.3 状态定义（State）

LangGraph依靠State在各个节点之间传递全局数据，包含：待抓取URL列表、原始文档内容、结构化政策信息、错误日志。

二、环境依赖安装

先执行依赖包安装

pip install langgraph langchain langchain-openai chromadb beautifulsoup4 requests pypdf2 pytesseract python-dotenv

注意：OCR识别需要本地安装Tesseract-OCR程序，并配置环境变量。Windows用户下载安装包，Linux执行sudo apt install tesseract-ocr。

三、完整代码实现

3.1 全局状态定义与基础初始化

新建power_policy_agent.py

import os
import requests
import pytesseract
from PyPDF2 import PdfReader
from bs4 import BeautifulSoup
from typing import TypedDict, Annotated, Sequence, Any
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import PromptTemplate
from dotenv import load_dotenv

load_dotenv()

# 配置大模型，支持通义千问、DeepSeek等兼容OpenAI接口模型
llm = ChatOpenAI(
model="deepseek-chat",
api_key=os.getenv("LLM_API_KEY"),
base_url=os.getenv("LLM_BASE_URL"),
temperature=0.1
)

# 向量数据库初始化
embedding = OpenAIEmbeddings(
model="text-embedding",
api_key=os.getenv("EMBED_KEY"),
base_url=os.getenv("EMBED_BASE")
)
vector_db = Chroma(persist_directory="./power_policy_db", embedding_function=embedding)

# ========== 定义全局状态 ==========
class AgentState(TypedDict):
url_list: Annotated[list[str], operator.add] # 需要抓取的政策网页链接
raw_documents: Annotated[list[dict], operator.add] # 原始文档数据 {url, content, doc_type}
structured_policy: Annotated[list[dict], operator.add] # 大模型抽取结构化政策
error_log: Annotated[list[str], operator.add] # 错误日志

3.2 Agent节点1：爬虫智能体（CrawlerAgent）

负责访问电力交易网站，抓取政策页面内容，区分普通网页、PDF附件链接。

def crawler_agent(state: AgentState) –> AgentState:
"""爬虫智能体：遍历url，获取网页内容，识别PDF附件"""
new_raw_docs = []
error = []
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
for url in state["url_list"]:
try:
resp = requests.get(url, headers=headers, timeout=15)
resp.raise_for_status()
# 判断是否为PDF文件
if url.endswith(".pdf") or "application/pdf" in resp.headers.get("Content-Type", ""):
new_raw_docs.append({
"source_url": url,
"doc_type": "pdf",
"binary_data": resp.content,
"text": ""
})
else:
soup = BeautifulSoup(resp.text, "html.parser")
# 清除网页标签，提取正文
text = soup.get_text(strip=True, separator="\\n")
new_raw_docs.append({
"source_url": url,
"doc_type": "webpage",
"binary_data": None,
"text": text
})
except Exception as e:
error.append(f"抓取失败 {url}: {str(e)}")
return {"raw_documents": new_raw_docs, "error_log": error}

3.3 Agent节点2：文档解析智能体（DocumentParserAgent）

分流处理三类数据：网页文本直接使用；PDF文本提取；图片附件调用OCR识别。

def document_parser_agent(state: AgentState) –> AgentState:
"""文档解析Agent：PDF解析 + OCR图片识别"""
processed_docs = []
error = []
for doc in state["raw_documents"]:
try:
if doc["doc_type"] == "webpage":
processed_docs.append(doc)
elif doc["doc_type"] == "pdf":
# PDF解析
from io import BytesIO
pdf_stream = BytesIO(doc["binary_data"])
reader = PdfReader(pdf_stream)
full_text = ""
for page in reader.pages:
page_text = page.extract_text()
if page_text:
full_text += page_text + "\\n"
doc["text"] = full_text
processed_docs.append(doc)
elif doc["doc_type"] == "image":
# OCR图片识别，适用于扫描版政策文件
from PIL import Image
from io import BytesIO
img = Image.open(BytesIO(doc["binary_data"]))
ocr_text = pytesseract.image_to_string(img, lang="chi_sim")
doc["text"] = ocr_text
processed_docs.append(doc)
except Exception as e:
error.append(f"文档解析失败 {doc['source_url']}: {str(e)}")
return {"raw_documents": processed_docs, "error_log": error}

3.4 Agent节点3：政策抽取智能体（PolicyExtractAgent）

利用大模型从非结构化文本提取电力政策结构化信息，也是本项目核心业务节点。

extract_prompt = PromptTemplate(
input_variables=["document_text"],
template="""
你是电力市场政策分析师，请阅读下面电力交易政策文档，严格输出JSON格式，不要额外解释。
文档内容：
{document_text}

输出JSON字段要求：
{{
"province": "政策所属省份",
"policy_name": "政策文件全称",
"effective_date": "生效时间",
"publish_date": "发布时间",
"core_rules": ["规则要点1","规则要点2"],
"target_subject": "适用市场主体（发电企业/售电公司/电力用户）",
"change_content": "相较于旧政策的变更内容，如果无对比填暂无",
"impact_analysis": "该政策对电力交易收益、报价策略带来的影响简述"
}}
只返回标准JSON，禁止增加markdown、注释。
"""
)

def policy_extract_agent(state: AgentState) –> AgentState:
"""大模型抽取政策结构化信息"""
structured_result = []
error = []
for doc in state["raw_documents"]:
text = doc["text"]
if len(text.strip()) < 20:
error.append(f"文档{doc['source_url']}有效文本过短，跳过抽取")
continue
try:
chain = extract_prompt | llm
resp = chain.invoke({"document_text": text[:12000]}) # 截断超长文本
json_str = resp.content.strip()
import json
policy_data = json.loads(json_str)
policy_data["source_url"] = doc["source_url"]
policy_data["original_text"] = text[:5000]
structured_result.append(policy_data)
except Exception as e:
error.append(f"政策抽取失败 {doc['source_url']}: {str(e)}")
return {"structured_policy": structured_result, "error_log": error}

3.5 Agent节点4：知识库入库智能体（KnowledgeBaseAgent）

将结构化政策存入向量库，构建电力交易知识库，支持后续RAG检索。

def knowledge_base_agent(state: AgentState) –> AgentState:
"""将结构化政策写入向量知识库"""
error = []
texts = []
metadatas = []
for policy in state["structured_policy"]:
content = f"""
省份：{policy['province']}
政策名称：{policy['policy_name']}
生效时间：{policy['effective_date']}
核心规则：{";".join(policy['core_rules'])}
影响评估：{policy['impact_analysis']}
"""
texts.append(content)
metadatas.append({
"source_url": policy["source_url"],
"province": policy["province"],
"publish_date": policy["publish_date"]
})
if texts:
vector_db.add_texts(texts=texts, metadatas=metadatas)
vector_db.persist()
return {"error_log": error}

3.6 LangGraph 构建工作流图，启动任务

def build_graph():
graph = StateGraph(AgentState)
# 注册节点
graph.add_node("crawler", crawler_agent)
graph.add_node("parser", document_parser_agent)
graph.add_node("extractor", policy_extract_agent)
graph.add_node("knowledge_store", knowledge_base_agent)
# 设置流转顺序
graph.set_entry_point("crawler")
graph.add_edge("crawler", "parser")
graph.add_edge("parser", "extractor")
graph.add_edge("extractor", "knowledge_store")
graph.add_edge("knowledge_store", END)
return graph.compile()

if __name__ == "__main__":
app = build_graph()
# 输入待抓取电力政策链接，替换为各省电力交易中心公告地址
init_input = {
"url_list": [
"https://example.com/jiangsu_power_policy.pdf",
"https://example.com/henan_trade_notice.html"
],
"raw_documents": [],
"structured_policy": [],
"error_log": []
}
result = app.invoke(init_input)
print("====任务执行完成====")
print("错误日志：")
for err in result["error_log"]:
print(err)
print(f"成功解析政策数量：{len(result['structured_policy'])}")

四、代码运行说明与工程优化方案

4.1 基础运行问题解决

4.2 生产环境优化方向

五、业务落地思考：电力政策Agent的价值

传统方案使用固定爬虫+正则表达式提取政策条款，缺点十分明显：电力政策文件格式不统一、扫描版PDF无法提取文本、各省文件行文规范差异巨大，正则很难适配。

基于LangGraph多智能体方案优势：

六、现存痛点与避坑指南

七、拓展延伸

本套多智能体框架不局限于电力政策场景，可以快速迁移至：新能源政策跟踪、储能行业文件解析、招投标公告抓取。结合LangGraph的条件分支、循环能力，还可以实现自主搜索Agent：当政策信息缺失时，智能体主动生成关键词再次检索网页，形成闭环自主调研。