王声声-技术文档

# 智能体项目：【报告仿写】技术路线总结

# 整体架构

这是一个基于LangChain框架的链式AI处理系统，主要用于电力行业报告的智能仿写生成或仅换数。系统采用链式流水线处理模式，将复杂的报告生成任务分解为多个独立的处理环节执行。

# 核心设计理念

  1. 模块化设计：每个chain负责特定的功能，通过RunnableLambda进行连接

  2. 数据流转：通过标准化的数据格式在chain之间传递信息

  3. 异步处理：支持并发执行提高效率

  4. 模拟数据支持：内置模拟数据用于测试和演示

# 主要Chain功能分析

# 1. get_meta_chain_by_ways - 元数据提取链

功能：提取原始段落的基准信息

  • 提取报告区域（如"安徽省"）

  • 提取报告周期类型（如"旬度"、"月度"）

  • 提取时间范围（如"2025年6月上旬"）

  • 获取数据库表结构信息

# 2. task1_chain - 指标提取链

功能：从原文段落中提取数据指标规范

  • 分析段落内容，识别需要查询的数据指标

  • 生成标准化的指标描述表格

  • 输出格式：markdown表格形式的指标清单

# 3. task2_chain - 智能问数生成链

功能：根据提取的指标生成智能问数自然语言问句

  • 将指标转换为可查询的自然语言问题

  • 生成结构化的查询清单

  • 为后续的数据查询做准备

# 4. run_elc_service - 智能问数执行链

功能：调用智能问数业务接口获取实际数据

  • 并发执行多个智能问数请求

  • 调用ElecAskReportService服务

  • 将查询结果转换为结构化数据

  • 支持异步并发处理提高效率

# 5. task3_chain - 仿写生成链

功能：使用获取的数据进行段落仿写

  • 结合原始段落模板和查询到的数据

  • 生成新的报告段落内容

  • 输出格式：包含仿写结果和结构化数据

# 6. task4_huanshu_chain - 换数生成链

功能：替换原文中的具体数值

  • 保持原文结构不变

  • 仅替换数据指标的具体数值

  • 适用于数据更新场景

# 主链配置

# 仿写方案主链（task_type=1）

main_chain = (  get_meta_chain_by_ways 
              | task1_chain 
              | map_task1_to_task2_output  
              | task2_chain  
              | run_elc_service_sync 
              | task3_chain  )

# 换数方案主链（task_type=2）

main_chain_huanshu = (  get_meta_chain_by_ways  
                      | task1_chain  
                      | map_task1_to_task2_output 
                      | task2_chain  
                      | run_elc_service_sync 
                      | task4_huanshu_chain  )

# 数据处理流程

  1. 输入处理：接收报告标题、段落内容、区域信息等

  2. 元数据提取：确定查询的基准信息

  3. 指标识别：从段落中提取需要查询的数据指标

  4. 查询生成：将指标转换为智能问数查询

  5. 数据获取：并发执行查询获取实际数据

  6. 内容生成：根据数据生成新的报告内容

  7. 结果输出：返回仿写/换数后的段落和结构化数据

# 技术特点

  • 可扩展性：通过chain组合支持不同的处理方案

  • 性能优化：支持异步并发处理

  • 错误处理：完善的异常捕获和日志记录

  • 数据溯源：保留完整的处理过程和数据来源

  • 模拟支持：内置模拟数据便于测试和演示

# run_elc_service方法中大模型调用流程详解

# 1. chain_template_devider_chain - 智能问数清单拆分链

功能：将task2生成的智能问数清单拆分为独立的查询单元

输入：task2生成的markdown表格格式的智能问数清单

指标名称 地区维度 时间维度 行业维度 口径 智能问数自然语言问句
电量同比 安徽省 2025年6月 制造业 售电 安徽省制造业2025年6月电量同比增速是多少？
电量环比 安徽省 2025年6月 制造业 售电 安徽省制造业2025年6月电量环比增速是多少？

输出：JSON数组格式的独立查询单元

[
"```markdown\n| 指标名称 | 地区维度 | 时间维度 | 行业维度 | 口径 | 智能问数自然语言问句 |\n|---------|---------|---------|---------|-----|-------------------|\n| 电量同比 | 安徽省 | 2025年6月 | 制造业 | 售电 | 安徽省制造业2025年6月电量同比增速是多少？ |\n```",

"```markdown\n| 指标名称 | 地区维度 | 时间维度 | 行业维度 | 口径 | 智能问数自然语言问句 |\n|---------|---------|---------|---------|-----|-------------------|\n| 电量环比 | 安徽省 | 2025年6月 | 制造业 | 售电 | 安徽省制造业2025年6月电量环比增速是多少？ |\n```"
]

大模型作用：

  • 解析markdown表格结构

  • 将多行指标拆分为独立的查询单元

  • 确保每个查询单元包含完整的表头信息

  • 输出标准化的JSON数组格式

# 2. ElecAskReportService.askForReport - 智能问数服务链

功能：对每个独立的查询单元执行完整的智能问数流程

内部调用的大模型链：

# 2.1 意图识别链 (input_chain)

功能：识别用户查询的意图类型

输入：自然语言查询（如"安徽省制造业2025年6月电量同比增速是多少？"）

输出：意图分类结果（1-5，对应不同的处理模式）

  • "1": 通用SQL生成

  • "2": 外贸企业部门电量分析

  • "3": 外贸企业电量分析

  • "4": 业扩电量分析

  • "5": 营业户数指标分析

# 2.2 SQL生成链 (根据意图选择对应的chain)

功能：根据自然语言查询生成对应的SQL语句

输入：自然语言查询 + 表结构信息

输出：标准化的SQL查询语句

SELECT total_eq_yoy, org_name, ind_cls_name, stat_date
FROM hs_dianliang_indicator_result_month 
WHERE is_deleted = 1 
AND (org_name LIKE '%安徽省%' OR org_short_name LIKE '%安徽省%')
AND stat_date = '2025-06'
AND ind_cls_name LIKE '%制造业%'

# 2.3 结构化结果处理链 (sql_chain_simplified_results)

功能：将SQL查询结果转换为结构化的markdown表格

输入：SQL查询结果 + 原始SQL + 表结构

输出：标准化的markdown表格

指标名称 地区维度 时间维度 行业维度 口径 数据值 对应表名 对应字段名 源表名 数据类型 加工算法逻辑 说明
电量同比 安徽省 2025年6月 制造业 售电 -12.5% hs_dianliang_indicator_result_month total_eq_yoy hs_dianliang_indicator_result_day 原始数据 聚合 同比增速为负

# 3. 并发处理机制

设计理念：为了提高效率，系统采用异步并发处理多个智能问数请求

实现方式：

# 并发执行所有智能问数请求

async def process_single_ask(askItem):

 """处理单个智能问数请求"""

 # 在线程池中执行同步的askForReport方法

result = await loop.run_in_executor(
   None, 
   lambda: self.elc_service.askForReport(askItem, stream=False, use_simplified_results=True)
  )
 return {
    "polished": result["polished"],
    "jsonArray": result.get("jsonArray", [])
  }

# 并发执行
tasks = [process_single_ask(askItem) for askItem in askList]
results = await asyncio.gather(*tasks, return_exceptions=True)

# 4. 数据流转和整合

数据收集：

  • 收集所有并发查询的结果

  • 提取结构化数据（markdown表格和JSON数组）

  • 处理异常情况，确保系统稳定性

数据整合：

# 合并所有结果
askResultList = [] # 保存所有查询的markdown表格结果
askJsonArrayList = [] # 保存所有查询的结构化JSON结果

# 转换为最终输出格式
result = {
  "jsonArray": markdownTable or askJsonArrayList, # 结构化JSON数组
  "paragraph_to_imitate": self.reportGenChainModel.paragraphStr, # 原始段落
  "table_data": askResultList, # 所有查询的markdown表格结果
}

# 5. 技术特点总结

多层大模型调用：

  1. 拆分层：将复杂查询拆分为独立单元

  2. 意图识别层：确定查询类型和处理策略

  3. SQL生成层：将自然语言转换为SQL

  4. 结构化处理层：将查询结果标准化

性能优化：

  • 异步并发处理多个查询

  • 线程池执行同步操作

  • 异常处理和容错机制

数据标准化：

  • 统一的markdown表格格式

  • 结构化的JSON数据输出

  • 完整的数据溯源信息

← 安徽智能体优化方案 →