环评报告写作Agent架构优化设计

概述

本文档针对环评报告写作的特殊场景（高合规性、复杂表格、专业计算、频繁用户交互），提出Agent架构的优化设计方案。

环评报告写作场景特征分析

1. 高合规性要求

法规引用严格性: 每个结论都需要明确的法律依据
标准适用准确性: 必须准确引用适用的环境质量标准、排放标准
审批风险: 错误的法规引用可能导致报告被退回或法律风险

2. 复杂表格处理

表格类型多样: 污染源清单、影响预测结果、监测计划、达标分析等
数据关联性强: 表格数据之间存在计算关系和逻辑依赖
格式要求严格: 表格格式需符合行业规范

3. 专业数值计算

计算类型: 排放量核算、影响预测、达标分析、噪声衰减等
公式复杂度高: 涉及高斯扩散模型、噪声衰减公式等专业模型
精度要求高: 计算结果直接影响评价结论

4. 频繁用户交互

数据确认: 基础数据需要用户确认（如污染源参数）
结论审核: 评价结论需要专家审核
迭代修正: 根据审查意见反复修改

5. 章节内容依赖

数据流动: 前面章节的数据被后续章节引用
逻辑连贯: 评价结论需要与前面的分析一致
版本同步: 修改前面章节时需要同步更新后续章节

6. 质量要求极高

零容错: 错误可能导致严重后果
可追溯性: 需要记录数据来源和计算过程
审查严格: 需要通过多轮专家审查

现有Skills能力盘点

系统已实现以下Skills，可直接用于环评报告写作：

1. CalculatorSkill (skill-calculator)

功能: 执行环评相关的数据计算 内置公式:

排放量计算: emission_rate = production * emission_factor
浓度计算: concentration = mass / volume
流量计算: flow_rate = velocity * area
粉尘排放量: dust_emission = area * dust_factor * time
废水排放量: wastewater = water_consumption * discharge_rate
噪声衰减: noise_attenuation = source_level - 20 * log10(distance / reference_distance)
高斯扩散模型（简化）

适用场景:

污染源强核算
环境影响预测
达标分析计算

2. RegulationSearcherSkill (skill-regulation-searcher)

功能: 搜索法规标准并生成引用格式特性:

混合检索（向量 + 关键词）
三层金字塔访问控制（L1公共/L2区域/L3项目）
自动生成标准引用格式
时效性排序（状态、级别、年份）

适用场景:

法规标准查询
合规性检查
引用格式生成

3. RetrieverSkill (skill-retriever)

功能: 从知识库检索相关内容特性:

向量语义检索
三层金字塔架构
查询增强
结果排序

适用场景:

历史案例检索
技术资料查询
相似项目参考

4. SlotReplacerSkill (skill-slot-replacer)

功能: 替换文档中的插槽变量特性:

单个/批量替换
格式保持
验证检查

适用场景:

项目信息填充
数据自动填充
模板实例化

5. TemplateInserterSkill (skill-template-inserter)

功能: 在编辑器指定位置插入模板内容特性:

位置控制（光标/末尾/指定位置）
格式转换（HTML/Markdown/Text）
插槽标记保留

适用场景:

章节模板插入
标准段落插入
表格模板插入

6. SummarizerSkill (skill-summarizer)

功能: 文本总结和关键词提取特性:

摘要生成
关键词提取
长度控制

适用场景:

章节摘要生成
报告概述编写
关键信息提取

7. ExporterSkill (skill-exporter)

功能: 导出报告为PDF或Word格式特性:

多格式导出（PDF/DOCX）
模板支持
样式配置
目录生成

适用场景:

报告导出
文档归档
成果交付

任务

⚠️ 计划制定需要额外时间

Agent设计模式适配性评估

针对环评报告写作场景，评估7种主流Agent设计模式：

1. ReAct模式（Reasoning and Acting）

原理: 推理和行动交替进行，每一步都先思考再行动

适配性评估: ⭐⭐⭐⭐

✅ 适合需要逐步推理的场景（如法规适用性判断）
✅ 可以在每一步记录推理过程，提高可追溯性
⚠️ 对于简单任务可能过于复杂
⚠️ 推理步骤过多会影响响应速度

推荐应用场景:

法规标准适用性分析
环境影响评价结论推导
合规性检查

2. Plan-and-Execute模式

原理: 先制定完整计划，再按计划执行

适配性评估: ⭐⭐⭐⭐⭐

✅ 非常适合环评报告的多步骤编写流程
✅ 可以提前识别章节依赖关系
✅ 便于用户审查和调整计划
✅ 支持并行执行独立 推荐应用场景:
完整章节编写
多章节协同生成
报告全流程编写

最适合环评场景！

3. Reflection模式（自我反思）

原理: 生成内容后进行自我验证和迭代改进

适配性评估: ⭐⭐⭐⭐⭐

✅ 非常适合高质量要求的环评报告
✅ 可以自动发现逻辑错误和数据不一致
✅ 减少人工审查工作量
✅ 提高输出质量
⚠️ 需要额外的验证步骤

推荐应用场景:

评价结论验证
数据一致性检查
法规引用验证

必须集成！

4. Human-in-the-Loop模式

原理: 在关键节点引入人工确认和反馈

适配性评估: ⭐⭐⭐⭐⭐

✅ 完美匹配环评报告的频繁交互需求
✅ 在关键数据和结论处设置确认点
✅ 支持专家审核和修正
✅ 降低错误风险
⚠️ 需要设计良好的交互界面

推荐应用场景:

基础数据确认
计算结果审核
评价结论审查
法规引用确认

必须集成！

5. Multi-Agent协作模式

原理: 多个专业Agent协同工作

适配性评估: ⭐⭐⭐⭐

✅ 适合环评报告的多专业协同需求
✅ 可以并行处理不同专业内容
✅ 每个Agent专注于特定领域
⚠️ 需要协调机制
⚠️ 实现复杂度较高

推荐应用场景:

多专业章节并行编写（大气、水、声、固废等）
复杂任务分解执行
专业知识整合

6. Tool-Calling模式

原理: Agent调用外部工具完成特定任务

适配性评估: ⭐⭐⭐⭐⭐

✅ 完美匹配环评报告的计算和查询需求
✅ 可以集成现有的Skills
✅ 支持调用专业计算引擎
✅ 易于扩展新工具
⚠️ 需要工具注册和管理机制

推荐应用场景:

调用CalculatorSkill进行计算
调用RegulationSearcherSkill查询法规
调用数据库查询项目信息
调用外部API获取数据

必须集成！

7. 混合模式

原理: 组合多种模式的优势

适配性评估: ⭐⭐⭐⭐⭐

✅ 最适合环评报告的复杂场景
✅ 可以根据任务类型选择最优模式
✅ 灵活性最高
⚠️ 设计和实现复杂度最高

推荐方案: Plan-and-Execute + Reflection + Human-in-the-Loop + Tool-Calling

推荐的Agent架构设计

基于以上分析，推荐采用混合模式，核心组合为：

Plan-and-Execute + Reflection + Human-in-the-Loop + Tool-Calling

架构图

用户请求（编写某章节）
    ↓
[Planning Phase] 规划阶段
    ├─ 分析任务需求
    ├─ 识别数据依赖
    ├─ 制定执行计划
    └─ 用户确认计划 ← Human-in-the-Loop
    ↓
[Execution Phase] 执行阶段
    ├─ Task 1: 数据收集
    │   ├─ 调用RetrieverSkill检索历史数据
    │   ├─ 调用RegulationSearcherSkill查询法规
    │   └─ 用户确认数据 ← Human-in-the-Loop
    ├─ Task 2: 数据计算
    │   ├─ 调用CalculatorSkill执行计算
    │   └─ 用户确认计算结果 ← Human-in-the-Loop
    ├─ Task 3: 内容生成
    │   ├─ 调用LLM生成文本
    │   ├─ 调用TemplateInserterSkill插入模板
    │   └─ 调用SlotReplacerSkill填充数据
    └─ Task 4: 表格生成
        └─ 生成复杂表格
    ↓
[Reflection Phase] 反思验证阶段
    ├─ 数据一致性检查
    ├─ 法规引用验证
    ├─ 逻辑流程验证
    ├─ 计算结果验证
    └─ 生成验证报告
    ↓
[Review Phase] 人工审查阶段
    ├─ 展示生成内容
    ├─ 展示验证报告
    ├─ 用户审查和修正 ← Human-in-the-Loop
    └─ 如需修改，返回执行阶段
    ↓
[Finalization Phase] 完成阶段
    ├─ 保存内容
    ├─ 记录执行日志
    └─ 返回结果

核心组件设计

1. PlanningAgent（规划Agent）

职责: 分析任务，制定执行计划

输入:

用户请求（如"编写第7章环境影响预测与评价"）
当前文档状态
项目上下文信息

输出:

任务分解列表（Task List）
数据依赖关系图
执行顺序建议
预估时间

关键能力:

识别章节依赖关系
识别需要的数据和工具
识别需要用户确认的节点

实现方式:

class PlanningAgent:
    async def plan(self, request: UserRequest, context: Context) -> ExecutionPlan:
        # 1. 分析任务类型
        task_type = self._analyze_task_type(request)
        
        # 2. 识别依赖
        dependencies = self._identify_dependencies(request, context)
        
        # 3. 分解任务
        tasks = self._decompose_tasks(request, dependencies)
        
        # 4. 识别确认点
        checkpoints = self._identify_checkpoints(tasks)
        
        # 5. 生成计划
        plan = ExecutionPlan(
            tasks=tasks,
            dependencies=dependencies,
            checkpoints=checkpoints,
            estimated_time=self._estimate_time(tasks)
        )
        
        return plan

2. ExecutionAgent（执行Agent）

职责: 按计划执行任务，调用Skills和工具

输入:

执行计划（ExecutionPlan）
任务参数

输出:

执行结果
中间数据
执行日志

关键能力:

任务调度（串行/并行）
Skills调用
状态管理
错误处理

实现方式:

class ExecutionAgent:
    def __init__(self):
        self.skill_registry = SkillRegistry()
        self.state_manager = StateManager()
    
    async def execute(self, plan: ExecutionPlan) -> ExecutionResult:
        results = []
        
        for task in plan.tasks:
            # 检查依赖是否满足
            if not self._check_dependencies(task, results):
                await self._wait_for_dependencies(task)
            
            # 执行任务
            if task.type == "skill_call":
                result = await self._execute_skill(task)
            elif task.type == "llm_generation":
                result = await self._execute_llm(task)
            elif task.type == "user_confirmation":
                result = await self._request_user_confirmation(task)
            
            # 保存结果
            results.append(result)
            self.state_manager.save_state(task.id, result)
        
        return ExecutionResult(results=results)
    
    async def _execute_skill(self, task: Task) -> SkillResult:
        skill = self.skill_registry.get(task.skill_name)
        return await skill.execute(**task.params)

3. ReflectionAgent（反思验证Agent）

职责: 验证生成内容的正确性和一致性

输入:

生成的内容
执行过程数据
验证规则

输出:

验证报告
问题列表
修正建议

验证维度:

数据一致性: 检查数据是否前后一致
法规引用: 验证法规引用是否正确
计算验证: 验证计算结果是否合理
逻辑流程: 检查逻辑是否连贯
格式规范: 检查格式是否符合要求

实现方式:

class ReflectionAgent:
    async def validate(self, content: str, context: Context) -> ValidationReport:
        issues = []
        
        # 1. 数据一致性检查
        data_issues = await self._check_data_consistency(content, context)
        issues.extend(data_issues)
        
        # 2. 法规引用验证
        regulation_issues = await self._validate_regulations(content)
        issues.extend(regulation_issues)
        
        # 3. 计算验证
        calculation_issues = await self._validate_calculations(content, context)
        issues.extend(calculation_issues)
        
        # 4. 逻辑流程检查
        logic_issues = await self._check_logic_flow(content, context)
        issues.extend(logic_issues)
        
        # 5. 格式规范检查
        format_issues = await self._check_format(content)
        issues.extend(format_issues)
        
        return ValidationReport(
            issues=issues,
            severity=self._calculate_severity(issues),
            suggestions=self._generate_suggestions(issues)
        )

4. InteractionManager（交互管理器）

职责: 管理与用户的交互，设置确认点

交互类型:

数据确认: 确认基础数据是否正确
计算审核: 审核计算结果
内容审查: 审查生成的文本内容
问题反馈: 处理验证发现的问题

实现方式:

class InteractionManager:
    async def request_confirmation(
        self, 
        checkpoint: Checkpoint, 
        data: Any
    ) -> ConfirmationResult:
        # 构建确认请求
        request = ConfirmationRequest(
            type=checkpoint.type,
            title=checkpoint.title,
            data=data,
            options=["确认", "修改", "跳过"]
        )
        
        # 发送到前端
        response = await self._send_to_frontend(request)
        
        # 处理用户响应
        if response.action == "修改":
            modified_data = response.modified_data
            return ConfirmationResult(confirmed=True, data=modified_data)
        elif response.action == "确认":
            return ConfirmationResult(confirmed=True, data=data)
        else:
            return ConfirmationResult(confirmed=False, data=data)

工作流程详细设计

场景示例：编写"第7章环境影响预测与评价"

Step 1: Planning Phase（规划阶段）

用户: "帮我编写第7章环境影响预测与评价"

PlanningAgent分析:
1. 识别章节类型: 影响预测章节
2. 识别依赖数据:
   - 污染源数据（来自第3章）
   - 环境现状数据（来自第5章）
   - 评价标准（需要查询法规库）
3. 分解任务:
   Task 1: 收集基础数据
   Task 2: 查询适用标准
   Task 3: 计算污染物排放量
   Task 4: 预测环境影响
   Task 5: 生成达标分析表格
   Task 6: 编写评价结论
4. 设置确认点:
   Checkpoint 1: 确认基础数据（Task 1后）
   Checkpoint 2: 确认计算结果（Task 3后）
   Checkpoint 3: 审查评价结论（Task 6后）

生成执行计划并展示给用户确认

Step 2: Execution Phase（执行阶段）

Task 1: 收集基础数据
├─ 调用RetrieverSkill检索第3章污染源数据
├─ 调用RetrieverSkill检索第5章环境现状数据
└─ 整理数据清单

Checkpoint 1: 用户确认数据
├─ 展示数据清单
├─ 用户确认或修正
└─ 保存确认后的数据

Task 2: 查询适用标准
├─ 调用RegulationSearcherSkill查询环境质量标准
├─ 调用RegulationSearcherSkill查询排放标准
└─ 生成标准引用列表

Task 3: 计算污染物排放量
├─ 调用CalculatorSkill计算废气排放量
├─ 调用CalculatorSkill计算废水排放量
├─ 调用CalculatorSkill计算固废产生量
└─ 生成排放量汇总表

Checkpoint 2: 用户确认计算结果
├─ 展示计算过程和结果
├─ 用户审核
└─ 保存确认后的结果

Task 4: 预测环境影响
├─ 调用CalculatorSkill（高斯扩散模型）预测大气影响
├─ 调用CalculatorSkill（噪声衰减）预测声环境影响
├─ 调用LLM生成影响分析文本
└─ 生成影响预测结果表

Task 5: 生成达标分析表格
├─ 对比预测值与标准值
├─ 生成达标分析表格
└─ 调用TemplateInserterSkill插入表格模板

Task 6: 编写评价结论
├─ 调用LLM基于预测结果生成结论
├─ 调用SlotReplacerSkill填充关键数据
└─ 生成完整章节内容

Step 3: Reflection Phase（反思验证阶段）

ReflectionAgent验证:

1. 数据一致性检查
   ✓ 污染源数据与第3章一致
   ✓ 环境现状数据与第5章一致
   ✗ 发现问题: 废水排放量与第3章不一致
   
2. 法规引用验证
   ✓ 《环境空气质量标准》GB 3095-2012 引用正确
   ✓ 《地表水环境质量标准》GB 3838-2002 引用正确
   
3. 计算验证
   ✓ 排放量计算公式正确
   ✓ 高斯扩散模型参数合理
   ⚠ 警告: 噪声衰减距离可能偏大
   
4. 逻辑流程检查
   ✓ 预测结果与评价结论逻辑一致
   
5. 格式规范检查
   ✓ 表格格式符合规范
   ✓ 章节结构完整

生成验证报告:
- 发现1个错误（数据不一致）
- 发现1个警告（参数可疑）
- 建议修正

Step 4: Review Phase（人工审查阶段）

展示给用户:
1. 生成的章节内容
2. 验证报告
3. 问题列表和修正建议

用户操作:
- 查看内容
- 查看验证报告
- 修正数据不一致问题
- 确认噪声衰减参数
- 批准最终内容

如果用户要求修改:
- 返回Execution Phase重新执行相关Task
- 重新进行Reflection验证

Step 5: Finalization Phase（完成阶段）

1. 保存章节内容到文档
2. 记录执行日志
3. 更新文档状态
4. 返回成功结果

Skills集成方案

Skills调用时机

阶段	Skills	调用场景
Planning	-	不直接调用Skills
Execution	RetrieverSkill	检索历史数据、案例
Execution	RegulationSearcherSkill	查询法规标准
Execution	CalculatorSkill	执行专业计算
Execution	TemplateInserterSkill	插入章节模板
Execution	SlotReplacerSkill	填充数据插槽
Execution	SummarizerSkill	生成摘要
Reflection	RegulationSearcherSkill	验证法规引用
Reflection	CalculatorSkill	验证计算结果
Finalization	ExporterSkill	导出报告

Skills扩展建议

基于环评场景需求，建议新增以下Skills：

1. TableGeneratorSkill（表格生成Skill）

功能: 生成复杂的环评表格特性:

支持多种表格类型（污染源清单、监测计划、达标分析等）
自动计算表格数据
支持表格格式验证

2. DataValidatorSkill（数据验证Skill）

功能: 验证数据的合理性和一致性特性:

数值范围检查
单位一致性检查
前后数据一致性检查
逻辑关系验证

3. CitationFormatterSkill（引用格式化Skill）

功能: 格式化法规标准引用特性:

自动生成标准引用格式
引用编号管理
引用列表生成

4. ChapterLinkerSkill（章节关联Skill）

功能: 管理章节间的数据依赖和引用特性:

追踪数据来源
自动更新引用
依赖关系可视化

状态管理设计

状态结构

class WritingState:
    """写作状态"""
    
    # 基本信息
    session_id: str
    user_id: str
    project_id: str
    document_id: str
    
    # 当前任务
    current_phase: str  # planning/execution/reflection/review/finalization
    current_task: Task | None
    
    # 执行计划
    plan: ExecutionPlan | None
    
    # 执行结果
    task_results: dict[str, TaskResult]
    
    # 中间数据
    collected_data: dict[str, Any]
    calculation_results: dict[str, Any]
    generated_content: dict[str, str]
    
    # 验证结果
    validation_report: ValidationReport | None
    
    # 用户交互
    pending_confirmations: list[Checkpoint]
    user_feedback: list[UserFeedback]
    
    # 上下文
    document_context: DocumentContext
    project_context: ProjectContext
    
    # 记忆
    conversation_history: list[Message]
    execution_history: list[ExecutionRecord]

状态持久化

class StateManager:
    """状态管理器"""
    
    def __init__(self, redis_client):
        self.redis = redis_client
        self.ttl = 86400  # 24小时
    
    async def save_state(self, session_id: str, state: WritingState):
        """保存状态"""
        key = f"writing:state:{session_id}"
        value = state.model_dump_json()
        await self.redis.setex(key, self.ttl, value)
    
    async def load_state(self, session_id: str) -> WritingState | None:
        """加载状态"""
        key = f"writing:state:{session_id}"
        value = await self.redis.get(key)
        if value:
            return WritingState.model_validate_json(value)
        return None
    
    async def update_state(self, session_id: str, updates: dict):
        """更新状态"""
        state = await self.load_state(session_id)
        if state:
            for key, value in updates.items():
                setattr(state, key, value)
            await self.save_state(session_id, state)

状态恢复

支持从任意阶段恢复执行：

class WritingOrchestrator:
    """写作编排器"""
    
    async def resume(self, session_id: str):
        """恢复执行"""
        # 加载状态
        state = await self.state_manager.load_state(session_id)
        if not state:
            raise ValueError("Session not found")
        
        # 根据当前阶段恢复
        if state.current_phase == "planning":
            return await self.planning_agent.plan(state)
        elif state.current_phase == "execution":
            return await self.execution_agent.execute(state)
        elif state.current_phase == "reflection":
            return await self.reflection_agent.validate(state)
        elif state.current_phase == "review":
            return await self.interaction_manager.request_review(state)
        else:
            raise ValueError(f"Unknown phase: {state.current_phase}")

用户交互设计

确认点类型

1. 数据确认（Data Confirmation）

时机: 收集基础数据后 展示内容:

数据清单
数据来源
数据值

用户操作:

确认
修改
添加备注

2. 计算审核（Calculation Review）

时机: 执行计算后 展示内容:

计算公式
输入参数
计算结果
计算过程

用户操作:

确认
修改参数
重新计算

3. 内容审查（Content Review）

时机: 生成内容后 展示内容:

生成的文本
验证报告
问题列表

用户操作:

批准
修改
重新生成

4. 问题反馈（Issue Feedback）

时机: 验证发现问题时 展示内容:

问题描述
严重程度
修正建议

用户操作:

接受建议
手动修正
忽略问题

交互界面设计

// 确认点组件
interface CheckpointComponent {
  type: 'data' | 'calculation' | 'content' | 'issue'
  title: string
  data: any
  validationReport?: ValidationReport
  actions: CheckpointAction[]
}

interface CheckpointAction {
  label: string
  type: 'confirm' | 'modify' | 'skip' | 'regenerate'
  handler: (data: any) => Promise<void>
}

// 使用示例
<CheckpointDialog
  type="calculation"
  title="确认计算结果"
  data={calculationResult}
  actions={[
    { label: '确认', type: 'confirm', handler: handleConfirm },
    { label: '修改参数', type: 'modify', handler: handleModify },
    { label: '重新计算', type: 'regenerate', handler: handleRecalculate }
  ]}
/>

关键技术挑战与解决方案

1. 复杂表格处理

挑战:

环评表格结构复杂，包含多层表头、合并单元格
表格数据之间存在计算关系
表格格式需要符合行业规范

解决方案:

方案A: 表格模板 + 数据填充

class TableGenerator:
    """表格生成器"""
    
    async def generate_table(
        self,
        template_id: str,
        data: dict[str, Any]
    ) -> Table:
        # 1. 加载表格模板
        template = await self.load_template(template_id)
        
        # 2. 计算派生数据
        computed_data = await self._compute_derived_data(data, template.formulas)
        
        # 3. 填充表格
        table = self._fill_table(template, {**data, **computed_data})
        
        # 4. 验证表格
        validation = self._validate_table(table, template.rules)
        
        return table

# 表格模板定义
pollution_source_template = {
    "id": "pollution_source_table",
    "name": "污染源清单表",
    "headers": [
        ["污染源", "污染物", "产生量", "削减量", "排放量", "排放方式"],
        ["", "", "(t/a)", "(t/a)", "(t/a)", ""]
    ],
    "formulas": {
        "排放量": "产生量 - 削减量"
    },
    "rules": {
        "排放量": {"min": 0, "type": "number"}
    }
}

方案B: LLM生成 + 结构化验证

async def generate_table_with_llm(
    self,
    description: str,
    data: dict[str, Any]
) -> Table:
    # 1. LLM生成表格结构
    prompt = f"""
    根据以下描述生成表格：
    {description}
    
    数据：
    {json.dumps(data, ensure_ascii=False)}
    
    返回JSON格式的表格结构。
    """
    
    response = await self.llm.ainvoke(prompt)
    table_json = json.loads(response.content)
    
    # 2. 验证表格结构
    table = Table.model_validate(table_json)
    
    # 3. 验证数据
    validation = self._validate_table_data(table)
    
    return table

推荐: 方案A（模板方式）更可靠，方案B作为补充

2. 专业计算集成

挑战:

计算公式复杂，涉及多个专业领域
需要保证计算精度和可追溯性
需要支持自定义公式

解决方案:

增强CalculatorSkill

class EnhancedCalculatorSkill(CalculatorSkill):
    """增强的计算技能"""
    
    async def execute_with_trace(
        self,
        formula: str,
        inputs: dict[str, float],
        **kwargs
    ) -> CalculationResult:
        # 1. 记录输入
        trace = CalculationTrace(
            formula=formula,
            inputs=inputs,
            timestamp=datetime.now()
        )
        
        # 2. 执行计算
        result = await super().execute(formula=formula, inputs=inputs, **kwargs)
        
        # 3. 记录输出
        trace.output = result.output
        trace.steps = self._extract_calculation_steps(formula, inputs)
        
        # 4. 保存追踪记录
        await self._save_trace(trace)
        
        return CalculationResult(
            value=result.output["result"],
            trace=trace,
            success=result.success
        )
    
    def _extract_calculation_steps(
        self,
        formula: str,
        inputs: dict[str, float]
    ) -> list[CalculationStep]:
        """提取计算步骤"""
        steps = []
        # 解析公式，生成计算步骤
        # 例如: "a * b + c" -> ["a * b = 50", "50 + c = 53"]
        return steps

集成外部计算引擎

class ExternalCalculationEngine:
    """外部计算引擎集成"""
    
    async def call_gaussian_model(
        self,
        params: GaussianModelParams
    ) -> GaussianModelResult:
        """调用高斯扩散模型"""
        # 调用外部Python库或API
        result = await self.gaussian_lib.calculate(params)
        return result
    
    async def call_noise_model(
        self,
        params: NoiseModelParams
    ) -> NoiseModelResult:
        """调用噪声预测模型"""
        result = await self.noise_lib.calculate(params)
        return result

3. 法规标准自动匹配

挑战:

法规标准数量庞大，更新频繁
需要准确判断适用性
需要生成标准引用格式

解决方案:

智能法规匹配

class IntelligentRegulationMatcher:
    """智能法规匹配器"""
    
    async def match_regulations(
        self,
        context: ProjectContext,
        requirement_type: str
    ) -> list[Regulation]:
        # 1. 提取匹配特征
        features = self._extract_features(context)
        # 特征: 行业类型、区域、污染物类型、项目规模等
        
        # 2. 向量检索候选法规
        candidates = await self.regulation_searcher.execute(
            keyword=requirement_type,
            top_k=20,
            project_context=context
        )
        
        # 3. LLM判断适用性
        applicable = []
        for reg in candidates:
            prompt = f"""
            判断以下法规是否适用于该项目：
            
            项目信息：
            - 行业: {context.industry}
            - 区域: {context.region}
            - 污染物: {context.pollutants}
            
            法规信息：
            - 名称: {reg.name}
            - 适用范围: {reg.scope}
            - 状态: {reg.status}
            
            返回JSON: {{"applicable": true/false, "reason": "原因"}}
            """
            
            response = await self.llm.ainvoke(prompt)
            result = json.loads(response.content)
            
            if result["applicable"]:
                applicable.append(reg)
        
        # 4. 按优先级排序
        sorted_regs = self._sort_by_priority(applicable)
        
        return sorted_regs

4. 长流程状态管理

挑战:

环评报告编写是长流程任务，可能跨越多天
需要支持中断和恢复
需要保持状态一致性

解决方案:

分布式状态管理

class DistributedStateManager:
    """分布式状态管理器"""
    
    def __init__(self, redis_client, db_client):
        self.redis = redis_client  # 热数据
        self.db = db_client  # 冷数据
    
    async def save_state(self, session_id: str, state: WritingState):
        # 1. 保存到Redis（热数据，快速访问）
        await self.redis.setex(
            f"state:{session_id}",
            3600,  # 1小时TTL
            state.model_dump_json()
        )
        
        # 2. 异步保存到数据库（冷数据，持久化）
        await self.db.execute(
            """
            INSERT INTO writing_states (session_id, state, updated_at)
            VALUES (:session_id, :state, :updated_at)
            ON CONFLICT (session_id) DO UPDATE SET
                state = :state,
                updated_at = :updated_at
            """,
            {
                "session_id": session_id,
                "state": state.model_dump_json(),
                "updated_at": datetime.now()
            }
        )
    
    async def load_state(self, session_id: str) -> WritingState | None:
        # 1. 尝试从Redis加载
        cached = await self.redis.get(f"state:{session_id}")
        if cached:
            return WritingState.model_validate_json(cached)
        
        # 2. 从数据库加载
        row = await self.db.fetch_one(
            "SELECT state FROM writing_states WHERE session_id = :session_id",
            {"session_id": session_id}
        )
        
        if row:
            state = WritingState.model_validate_json(row["state"])
            # 回写到Redis
            await self.redis.setex(
                f"state:{session_id}",
                3600,
                state.model_dump_json()
            )
            return state
        
        return None

状态快照和版本管理

class StateVersionManager:
    """状态版本管理器"""
    
    async def create_snapshot(
        self,
        session_id: str,
        state: WritingState,
        description: str
    ) -> str:
        """创建状态快照"""
        snapshot_id = f"{session_id}_{datetime.now().isoformat()}"
        
        await self.db.execute(
            """
            INSERT INTO state_snapshots (
                snapshot_id, session_id, state, description, created_at
            ) VALUES (:snapshot_id, :session_id, :state, :description, :created_at)
            """,
            {
                "snapshot_id": snapshot_id,
                "session_id": session_id,
                "state": state.model_dump_json(),
                "description": description,
                "created_at": datetime.now()
            }
        )
        
        return snapshot_id
    
    async def restore_snapshot(
        self,
        snapshot_id: str
    ) -> WritingState:
        """恢复快照"""
        row = await self.db.fetch_one(
            "SELECT state FROM state_snapshots WHERE snapshot_id = :snapshot_id",
            {"snapshot_id": snapshot_id}
        )
        
        if not row:
            raise ValueError(f"Snapshot not found: {snapshot_id}")
        
        return WritingState.model_validate_json(row["state"])

5. 性能优化

挑战:

环评报告编写涉及多次LLM调用，响应时间长
需要在保证质量的前提下提升速度

解决方案:

并行执行独立任务

class ParallelExecutor:
    """并行执行器"""
    
    async def execute_parallel(
        self,
        tasks: list[Task]
    ) -> list[TaskResult]:
        # 1. 分析任务依赖
        task_groups = self._group_by_dependencies(tasks)
        
        # 2. 按组执行
        all_results = []
        for group in task_groups:
            # 组内并行执行
            results = await asyncio.gather(*[
                self._execute_task(task)
                for task in group
            ])
            all_results.extend(results)
        
        return all_results
    
    def _group_by_dependencies(
        self,
        tasks: list[Task]
    ) -> list[list[Task]]:
        """按依赖关系分组"""
        groups = []
        remaining = tasks.copy()
        completed = set()
        
        while remaining:
            # 找出所有依赖已满足的任务
            ready = [
                task for task in remaining
                if all(dep in completed for dep in task.dependencies)
            ]
            
            if not ready:
                raise ValueError("Circular dependency detected")
            
            groups.append(ready)
            for task in ready:
                remaining.remove(task)
                completed.add(task.id)
        
        return groups

缓存和预加载

class CacheManager:
    """缓存管理器"""
    
    async def get_or_compute(
        self,
        key: str,
        compute_fn: Callable,
        ttl: int = 3600
    ) -> Any:
        """获取缓存或计算"""
        # 1. 尝试从缓存获取
        cached = await self.redis.get(f"cache:{key}")
        if cached:
            return json.loads(cached)
        
        # 2. 计算结果
        result = await compute_fn()
        
        # 3. 保存到缓存
        await self.redis.setex(
            f"cache:{key}",
            ttl,
            json.dumps(result)
        )
        
        return result
    
    async def preload_common_data(self, project_id: str):
        """预加载常用数据"""
        # 预加载项目信息
        await self.get_or_compute(
            f"project:{project_id}",
            lambda: self.db.get_project(project_id)
        )
        
        # 预加载常用法规
        await self.get_or_compute(
            f"regulations:common",
            lambda: self.regulation_searcher.get_common_regulations()
        )

流式响应

async def stream_generation(
    self,
    request: GenerationRequest
) -> AsyncIterator[str]:
    """流式生成内容"""
    async for chunk in self.llm.astream(request.prompt):
        # 实时返回生成的内容
        yield chunk.content
        
        # 同时进行后台验证
        if len(chunk.content) > 100:
            asyncio.create_task(
                self.reflection_agent.validate_partial(chunk.content)
            )

6. 用户体验优化

挑战:

频繁的用户确认可能影响体验
需要在质量和效率之间平衡

解决方案:

智能确认点

class SmartCheckpointManager:
    """智能确认点管理器"""
    
    def should_request_confirmation(
        self,
        checkpoint: Checkpoint,
        context: Context
    ) -> bool:
        """判断是否需要用户确认"""
        # 1. 高风险操作必须确认
        if checkpoint.risk_level == "high":
            return True
        
        # 2. 数据变化大时需要确认
        if self._has_significant_change(checkpoint, context):
            return True
        
        # 3. 验证发现问题时需要确认
        if checkpoint.validation_issues:
            return True
        
        # 4. 用户偏好设置
        user_prefs = context.user_preferences
        if user_prefs.confirmation_mode == "minimal":
            return False
        elif user_prefs.confirmation_mode == "all":
            return True
        
        # 5. 默认：中等风险需要确认
        return checkpoint.risk_level == "medium"

批量确认

async def batch_confirmation(
    self,
    checkpoints: list[Checkpoint]
) -> dict[str, ConfirmationResult]:
    """批量确认"""
    # 将多个确认点合并展示
    request = BatchConfirmationRequest(
        title="请确认以下内容",
        items=[
            {
                "id": cp.id,
                "title": cp.title,
                "data": cp.data,
                "default_action": "confirm"
            }
            for cp in checkpoints
        ]
    )
    
    response = await self.interaction_manager.request_batch_confirmation(request)
    
    return response.results

7. 可扩展性设计

挑战:

需要支持新的报告类型和章节模板
需要支持新的计算公式和验证规则

解决方案:

插件化架构

class PluginManager:
    """插件管理器"""
    
    def __init__(self):
        self.plugins: dict[str, Plugin] = {}
    
    def register_plugin(self, plugin: Plugin):
        """注册插件"""
        self.plugins[plugin.name] = plugin
        logger.info(f"Registered plugin: {plugin.name}")
    
    def get_plugin(self, name: str) -> Plugin | None:
        """获取插件"""
        return self.plugins.get(name)

# 插件示例：新的报告类型
class WaterEIAPlugin(Plugin):
    """水利水电环评插件"""
    
    name = "water-eia"
    
    def get_chapter_templates(self) -> list[ChapterTemplate]:
        """提供章节模板"""
        return [
            ChapterTemplate(
                id="water_eia_chapter_7",
                name="水环境影响评价",
                sections=[...]
            )
        ]
    
    def get_calculation_formulas(self) -> dict[str, Formula]:
        """提供计算公式"""
        return {
            "water_balance": Formula(
                name="水平衡计算",
                formula="input - output - loss",
                params=["input", "output", "loss"]
            )
        }
    
    def get_validation_rules(self) -> list[ValidationRule]:
        """提供验证规则"""
        return [
            ValidationRule(
                name="water_quality_check",
                description="检查水质指标是否符合标准",
                validator=self._validate_water_quality
            )
        ]

从当前架构到优化架构的迁移路径

迁移策略：渐进式演进

采用渐进式迁移策略，分阶段实施，降低风险：

阶段1：基础设施准备（2周）

目标: 搭建新架构的基础设施

任务:

实现状态管理器（StateManager）
实现交互管理器（InteractionManager）
扩展现有Skills（增加追踪、验证功能）
搭建测试环境

验收标准:

StateManager可以保存和恢复状态
InteractionManager可以处理用户确认
所有Skills支持执行追踪

风险: 低

阶段2：核心Agent实现（3周）

目标: 实现核心的Planning、Execution、Reflection Agent

任务:

实现PlanningAgent
- 任务分解逻辑
- 依赖关系分析
- 确认点识别
实现ExecutionAgent
- 任务调度器
- Skills调用封装
- 并行执行支持
实现ReflectionAgent
- 数据一致性检查
- 法规引用验证
- 计算结果验证

验收标准:

PlanningAgent可以生成合理的执行计划
ExecutionAgent可以按计划执行任务
ReflectionAgent可以发现常见问题

风险: 中

阶段3：试点章节迁移（2周）

目标: 选择1-2个章节试点新架构

推荐试点章节:

第7章（环境影响预测与评价）- 复杂度高，最能体现新架构优势
第8章（环境保护措施）- 相对简单，降低风险

任务:

为试点章节配置执行计划模板
实现章节特定的验证规则
设计用户交互流程
进行内部测试

验收标准:

试点章节可以完整生成
验证机制有效发现问题
用户反馈积极

风险: 中

阶段4：全面推广（4周）

目标: 将新架构推广到所有章节

任务:

为所有章节配置执行计划
完善验证规则库
优化性能（并行执行、缓存）
用户培训

验收标准:

所有章节支持新架构
性能满足要求（响应时间 < 30秒）
用户满意度 > 80%

风险: 中

阶段5：高级特性（持续）

目标: 持续优化和增强

任务:

实现智能确认点
实现批量确认
实现状态快照和版本管理
实现插件化架构
集成外部计算引擎

验收标准:

用户体验持续改善
系统可扩展性增强

风险: 低

兼容性方案

在迁移期间，新旧架构共存：

class HybridWritingService:
    """混合写作服务"""
    
    def __init__(self):
        self.legacy_service = LegacyWritingService()
        self.new_service = NewWritingService()
        self.feature_flags = FeatureFlags()
    
    async def handle_request(self, request: WritingRequest) -> WritingResponse:
        # 根据特性开关选择服务
        if self.feature_flags.is_enabled("new_architecture", request.user_id):
            # 使用新架构
            return await self.new_service.handle(request)
        else:
            # 使用旧架构
            return await self.legacy_service.handle(request)

回滚方案

如果新架构出现问题，可以快速回滚：

class RollbackManager:
    """回滚管理器"""
    
    async def rollback_to_legacy(self, session_id: str):
        """回滚到旧架构"""
        # 1. 加载新架构的状态
        new_state = await self.new_state_manager.load_state(session_id)
        
        # 2. 转换为旧架构的状态
        legacy_state = self._convert_to_legacy_state(new_state)
        
        # 3. 保存到旧架构
        await self.legacy_state_manager.save_state(session_id, legacy_state)
        
        # 4. 切换特性开关
        await self.feature_flags.disable("new_architecture", session_id)

实施建议

1. 团队组织

角色分工:

架构师: 负责整体架构设计和技术决策
后端开发: 实现Agent和Skills
前端开发: 实现交互界面
测试工程师: 编写测试用例，进行质量保证
领域专家: 提供环评专业知识，验证输出质量

2. 开发流程

迭代周期: 2周一个迭代

每个迭代:

需求评审（1天）
设计评审（1天）
开发实现（7天）
测试验证（2天）
演示反馈（1天）

3. 质量保证

测试策略:

单元测试: 覆盖率 > 80%
集成测试: 覆盖主要流程
端到端测试: 模拟真实场景
性能测试: 响应时间、并发能力
用户验收测试: 真实用户试用

代码审查:

所有代码必须经过审查
关键模块需要架构师审查

4. 监控和运维

监控指标:

响应时间
成功率
错误率
用户满意度
Skills调用次数
验证发现的问题数

告警机制:

响应时间 > 60秒
错误率 > 5%
验证失败率 > 20%

5. 文档和培训

文档:

架构设计文档
API文档
用户手册
运维手册

培训:

开发团队培训（技术实现）
用户培训（使用方法）
运维培训（监控和故障处理）

预期收益

1. 质量提升

错误率降低: 通过Reflection验证，预计错误率降低 60%
合规性提高: 通过法规引用验证，合规性提高 80%
一致性改善: 通过数据一致性检查，一致性问题减少 70%

2. 效率提升

编写速度: 通过并行执行和缓存，编写速度提升 40%
修改效率: 通过状态管理和版本控制，修改效率提升 50%
审查效率: 通过自动验证，人工审查工作量减少 50%

3. 用户体验

交互体验: 通过智能确认点，减少不必要的打扰
可追溯性: 通过执行追踪，提供完整的数据来源和计算过程
可恢复性: 通过状态管理，支持随时中断和恢复

4. 可维护性

代码质量: 通过模块化设计，代码更清晰
可扩展性: 通过插件化架构，易于扩展新功能
可测试性: 通过依赖注入，易于编写测试

总结

针对环评报告写作的特殊场景，推荐采用Plan-and-Execute + Reflection + Human-in-the-Loop + Tool-Calling的混合Agent架构：

核心优势

Plan-and-Execute: 适合多步骤、有依赖关系的复杂任务
Reflection: 自动验证，提高输出质量
Human-in-the-Loop: 在关键节点引入人工确认，降低风险
Tool-Calling: 充分利用现有Skills，支持专业计算

关键设计

四阶段工作流: Planning → Execution → Reflection → Review
状态管理: 支持长流程、可中断、可恢复
Skills集成: 充分利用现有的7个Skills
智能交互: 在关键节点设置确认点，平衡质量和效率

实施路径

渐进式迁移: 分5个阶段，降低风险
试点先行: 先在1-2个章节试点
新旧共存: 通过特性开关控制
可回滚: 出现问题可快速回滚

预期收益

错误率降低 60%
编写速度提升 40%
审查工作量减少 50%
用户满意度显著提升

这个架构设计充分考虑了环评报告写作的特殊性，在保证质量的前提下，提升效率和用户体验。

FilesExpand file tree

design-optimization.md

Latest commit

History

design-optimization.md

File metadata and controls

环评报告写作Agent架构优化设计

概述

环评报告写作场景特征分析

1. 高合规性要求

2. 复杂表格处理

3. 专业数值计算

4. 频繁用户交互

5. 章节内容依赖

6. 质量要求极高

现有Skills能力盘点

1. CalculatorSkill (skill-calculator)

2. RegulationSearcherSkill (skill-regulation-searcher)

3. RetrieverSkill (skill-retriever)

4. SlotReplacerSkill (skill-slot-replacer)

5. TemplateInserterSkill (skill-template-inserter)

6. SummarizerSkill (skill-summarizer)

7. ExporterSkill (skill-exporter)

Agent设计模式适配性评估

1. ReAct模式（Reasoning and Acting）

2. Plan-and-Execute模式

3. Reflection模式（自我反思）

4. Human-in-the-Loop模式

5. Multi-Agent协作模式

6. Tool-Calling模式

7. 混合模式

推荐的Agent架构设计

架构图

核心组件设计

1. PlanningAgent（规划Agent）

2. ExecutionAgent（执行Agent）

3. ReflectionAgent（反思验证Agent）

4. InteractionManager（交互管理器）

工作流程详细设计

场景示例：编写"第7章 环境影响预测与评价"

Skills集成方案

Skills调用时机

Skills扩展建议

1. TableGeneratorSkill（表格生成Skill）

2. DataValidatorSkill（数据验证Skill）

3. CitationFormatterSkill（引用格式化Skill）

4. ChapterLinkerSkill（章节关联Skill）

状态管理设计

状态结构

状态持久化

状态恢复

用户交互设计

确认点类型

1. 数据确认（Data Confirmation）

2. 计算审核（Calculation Review）

3. 内容审查（Content Review）

4. 问题反馈（Issue Feedback）

交互界面设计

关键技术挑战与解决方案

1. 复杂表格处理

方案A: 表格模板 + 数据填充

方案B: LLM生成 + 结构化验证

2. 专业计算集成

增强CalculatorSkill

集成外部计算引擎

3. 法规标准自动匹配

智能法规匹配

4. 长流程状态管理

分布式状态管理

状态快照和版本管理

5. 性能优化

并行执行独立任务

缓存和预加载

流式响应

6. 用户体验优化

智能确认点

批量确认

7. 可扩展性设计

插件化架构

从当前架构到优化架构的迁移路径

场景示例：编写"第7章环境影响预测与评价"