diff --git a/src/bidmaster/agents/__init__.py b/src/bidmaster/agents/__init__.py index 28d1f01..75553e4 100644 --- a/src/bidmaster/agents/__init__.py +++ b/src/bidmaster/agents/__init__.py @@ -2,5 +2,15 @@ from .analysis import AnalysisAgent from .toc_generator import TocGeneratorAgent +from .builders import TocAgentBuilder +from .builders.toc_builder import TocAgent +from .interaction import InteractionHandler, InteractionMode -__all__ = ["AnalysisAgent", "TocGeneratorAgent"] \ No newline at end of file +__all__ = [ + "AnalysisAgent", + "TocGeneratorAgent", + "TocAgentBuilder", + "TocAgent", + "InteractionHandler", + "InteractionMode" +] \ No newline at end of file diff --git a/src/bidmaster/agents/base.py b/src/bidmaster/agents/base.py new file mode 100644 index 0000000..e45d05f --- /dev/null +++ b/src/bidmaster/agents/base.py @@ -0,0 +1,276 @@ +"""Agent基础架构 + +提供Agent构建器和基础Agent类。 +""" + +import logging +from typing import List, Dict, Any, Optional, Callable, Tuple +from langgraph.graph import StateGraph, END + +from ..nodes.base import BaseNode, NodeContext + +logger = logging.getLogger(__name__) + + +class AgentBuilder: + """Agent构建器基类 + + 用于组装节点形成完整的Agent工作流。 + """ + + def __init__(self, interaction_handler: Optional[Callable] = None): + """初始化构建器 + + Args: + interaction_handler: 交互处理函数 + """ + self.nodes: List[BaseNode] = [] + self.edges: List[Tuple[str, str]] = [] + self.conditional_edges: Dict[str, Tuple[Callable, Dict[str, str]]] = {} + self.entry_point: Optional[str] = None + self.interaction_handler = interaction_handler + self.graph: Optional[StateGraph] = None + + def add_node(self, node: BaseNode) -> 'AgentBuilder': + """添加节点 + + Args: + node: 要添加的节点 + + Returns: + 自身，支持链式调用 + """ + self.nodes.append(node) + logger.debug(f"添加节点: {node.name}") + return self + + def add_edge(self, from_node: str, to_node: str) -> 'AgentBuilder': + """添加无条件边 + + Args: + from_node: 源节点名称 + to_node: 目标节点名称 + + Returns: + 自身，支持链式调用 + """ + self.edges.append((from_node, to_node)) + logger.debug(f"添加边: {from_node} -> {to_node}") + return self + + def add_conditional_edge(self, + from_node: str, + condition: Callable[[Dict[str, Any]], str], + routes: Dict[str, str]) -> 'AgentBuilder': + """添加条件边 + + Args: + from_node: 源节点名称 + condition: 条件函数，返回路由键 + routes: 路由映射 {条件结果: 目标节点} + + Returns: + 自身，支持链式调用 + """ + self.conditional_edges[from_node] = (condition, routes) + logger.debug(f"添加条件边: {from_node} -> {routes}") + return self + + def set_entry(self, node_name: str) -> 'AgentBuilder': + """设置入口节点 + + Args: + node_name: 入口节点名称 + + Returns: + 自身，支持链式调用 + """ + self.entry_point = node_name + logger.debug(f"设置入口点: {node_name}") + return self + + def build(self) -> StateGraph: + """构建LangGraph工作流 + + Returns: + 编译后的StateGraph实例 + + Raises: + ValueError: 配置不完整时抛出异常 + """ + if not self.nodes: + raise ValueError("至少需要一个节点") + + if not self.entry_point: + raise ValueError("必须设置入口点") + + logger.info(f"开始构建Agent，节点数: {len(self.nodes)}") + + # 创建工作流 + workflow = StateGraph(dict) # 使用通用dict作为状态类型 + + # 创建执行上下文 + context = NodeContext(self.interaction_handler) + + # 注册所有节点 + for node in self.nodes: + # 包装节点函数，注入context + def make_node_func(n: BaseNode) -> Callable: + def node_func(state: Dict[str, Any]) -> Dict[str, Any]: + try: + n._log_start() + result = n.execute(state, context) + n._log_success() + return result + except Exception as e: + n._log_error(e) + # 设置错误状态，但不中断流程（遵循立即失败原则） + raise e + + return node_func + + workflow.add_node(node.name, make_node_func(node)) + + # 设置入口点 + workflow.set_entry_point(self.entry_point) + + # 添加无条件边 + for from_node, to_node in self.edges: + if to_node == "END": + workflow.add_edge(from_node, END) + else: + workflow.add_edge(from_node, to_node) + + # 添加条件边 + for from_node, (condition, routes) in self.conditional_edges.items(): + # 处理END路由 + processed_routes = {} + for key, target in routes.items(): + processed_routes[key] = END if target == "END" else target + + workflow.add_conditional_edges(from_node, condition, processed_routes) + + # 编译工作流 + self.graph = workflow.compile() + logger.info("Agent构建完成") + + return self.graph + + def get_node_by_name(self, name: str) -> Optional[BaseNode]: + """根据名称获取节点 + + Args: + name: 节点名称 + + Returns: + 节点实例或None + """ + for node in self.nodes: + if node.name == name: + return node + return None + + def list_nodes(self) -> List[str]: + """列出所有节点名称 + + Returns: + 节点名称列表 + """ + return [node.name for node in self.nodes] + + def validate_configuration(self) -> List[str]: + """验证配置完整性 + + Returns: + 错误信息列表，空列表表示验证通过 + """ + errors = [] + + if not self.nodes: + errors.append("缺少节点") + + if not self.entry_point: + errors.append("缺少入口点") + elif self.entry_point not in self.list_nodes(): + errors.append(f"入口点节点不存在: {self.entry_point}") + + # 验证边的有效性 + node_names = set(self.list_nodes()) + for from_node, to_node in self.edges: + if from_node not in node_names: + errors.append(f"边的源节点不存在: {from_node}") + if to_node != "END" and to_node not in node_names: + errors.append(f"边的目标节点不存在: {to_node}") + + # 验证条件边 + for from_node, (_, routes) in self.conditional_edges.items(): + if from_node not in node_names: + errors.append(f"条件边的源节点不存在: {from_node}") + for target in routes.values(): + if target != "END" and target not in node_names: + errors.append(f"条件边的目标节点不存在: {target}") + + return errors + + +class BaseAgent: + """Agent基类 + + 提供Agent的通用功能。 + """ + + def __init__(self, builder: AgentBuilder): + """初始化Agent + + Args: + builder: 构建器实例 + """ + self.builder = builder + self.graph = builder.build() + + async def execute(self, initial_state: Dict[str, Any]) -> Dict[str, Any]: + """异步执行Agent + + Args: + initial_state: 初始状态 + + Returns: + 最终状态 + """ + logger.info(f"开始执行Agent，初始状态keys: {list(initial_state.keys())}") + + try: + final_state = await self.graph.ainvoke(initial_state) + logger.info("Agent执行完成") + return final_state + except Exception as e: + logger.error(f"Agent执行失败: {e}") + raise + + def execute_sync(self, initial_state: Dict[str, Any]) -> Dict[str, Any]: + """同步执行Agent + + Args: + initial_state: 初始状态 + + Returns: + 最终状态 + """ + import asyncio + return asyncio.run(self.execute(initial_state)) + + def get_node_names(self) -> List[str]: + """获取所有节点名称 + + Returns: + 节点名称列表 + """ + return self.builder.list_nodes() + + def validate(self) -> List[str]: + """验证Agent配置 + + Returns: + 错误信息列表 + """ + return self.builder.validate_configuration() \ No newline at end of file diff --git a/src/bidmaster/agents/builders/__init__.py b/src/bidmaster/agents/builders/__init__.py new file mode 100644 index 0000000..26eebbf --- /dev/null +++ b/src/bidmaster/agents/builders/__init__.py @@ -0,0 +1,10 @@ +"""Agent构建器模块 + +包含各种Agent的构建器实现。 +""" + +from .toc_builder import TocAgentBuilder + +__all__ = [ + "TocAgentBuilder" +] \ No newline at end of file diff --git a/src/bidmaster/agents/builders/toc_builder.py b/src/bidmaster/agents/builders/toc_builder.py new file mode 100644 index 0000000..b999f17 --- /dev/null +++ b/src/bidmaster/agents/builders/toc_builder.py @@ -0,0 +1,202 @@ +"""目录生成Agent构建器 + +用于构建TocGenerator Agent的完整工作流。 +""" + +import logging +from typing import Dict, Any, Optional, Callable + +from langgraph.graph import END + +from ..base import AgentBuilder, BaseAgent +from ...nodes.toc import ( + GroupCriteriaNode, + GenerateFirstLevelNode, + GenerateSubChaptersNode, + ReviewStructureNode, + FinalizeChaptersNode +) + +logger = logging.getLogger(__name__) + + +def should_continue(state: Dict[str, Any]) -> str: + """判断是否继续处理 + + Args: + state: 当前状态 + + Returns: + "continue" 或 "end" + """ + if state.get("error") or not state.get("should_continue", True): + return "end" + return "continue" + + +class TocAgentBuilder(AgentBuilder): + """目录生成Agent构建器""" + + @classmethod + def create(cls, interaction_handler: Optional[Callable] = None) -> 'TocAgentBuilder': + """创建预配置的TocAgent构建器 + + Args: + interaction_handler: 交互处理函数 + + Returns: + 配置完成的构建器实例 + """ + builder = cls(interaction_handler) + + # 添加所有节点 + builder.add_node(GroupCriteriaNode()) \ + .add_node(GenerateFirstLevelNode()) \ + .add_node(GenerateSubChaptersNode()) \ + .add_node(ReviewStructureNode()) \ + .add_node(FinalizeChaptersNode()) + + # 设置入口点 + builder.set_entry("group_criteria") + + # 配置工作流程 + builder._configure_workflow() + + logger.info("TocAgent构建器创建完成") + return builder + + def _configure_workflow(self) -> None: + """配置工作流程""" + # 添加条件边 - 每个节点都检查是否应该继续 + self.add_conditional_edge( + "group_criteria", + should_continue, + {"continue": "generate_first_level", "end": "END"} + ) + + self.add_conditional_edge( + "generate_first_level", + should_continue, + {"continue": "generate_sub_chapters", "end": "END"} + ) + + self.add_conditional_edge( + "generate_sub_chapters", + should_continue, + {"continue": "review_structure", "end": "END"} + ) + + self.add_conditional_edge( + "review_structure", + should_continue, + {"continue": "finalize_chapters", "end": "END"} + ) + + # 最后一个节点直接结束 + self.add_edge("finalize_chapters", "END") + + +class TocAgent(BaseAgent): + """目录生成Agent + + 封装了目录生成的完整工作流程。 + """ + + def __init__(self, interaction_handler: Optional[Callable] = None): + """初始化TocAgent + + Args: + interaction_handler: 交互处理函数 + """ + builder = TocAgentBuilder.create(interaction_handler) + super().__init__(builder) + + @classmethod + def create_with_handler(cls, interaction_handler: Callable) -> 'TocAgent': + """使用指定的交互处理器创建Agent + + Args: + interaction_handler: 交互处理函数 + + Returns: + TocAgent实例 + """ + return cls(interaction_handler) + + @classmethod + def create_silent(cls) -> 'TocAgent': + """创建静默模式的Agent（使用默认值） + + Returns: + 静默模式的TocAgent实例 + """ + from ..interaction import InteractionHandler, InteractionMode + handler = InteractionHandler(mode=InteractionMode.SILENT) + return cls(handler) + + @classmethod + def create_programmatic(cls, presets: Dict[str, Any]) -> 'TocAgent': + """创建程序化模式的Agent（使用预设值） + + Args: + presets: 预设值字典 + + Returns: + 程序化模式的TocAgent实例 + """ + from ..interaction import InteractionHandler, InteractionMode + handler = InteractionHandler(mode=InteractionMode.PROGRAMMATIC, presets=presets) + return cls(handler) + + async def generate_toc(self, + technical_criteria: list, + generation_mode: Optional[str] = None, + template_file: Optional[str] = None) -> Dict[str, Any]: + """生成目录结构 + + Args: + technical_criteria: 技术评分项列表 + generation_mode: 生成模式（ai/template），可选 + template_file: 模板文件路径，可选 + + Returns: + 包含生成结果的状态字典 + """ + initial_state = { + "technical_criteria": technical_criteria, + "should_continue": True, + "current_step": "", + "category_groups": {}, + "preliminary_chapters": [], + "structure_review": {}, + "final_chapters": [], + "error": "", + "warnings": [] + } + + # 如果提供了预设参数，添加到状态中 + if generation_mode: + initial_state["preset_generation_mode"] = generation_mode + if template_file: + initial_state["preset_template_file"] = template_file + + return await self.execute(initial_state) + + def generate_toc_sync(self, + technical_criteria: list, + generation_mode: Optional[str] = None, + template_file: Optional[str] = None) -> Dict[str, Any]: + """同步生成目录结构 + + Args: + technical_criteria: 技术评分项列表 + generation_mode: 生成模式（ai/template），可选 + template_file: 模板文件路径，可选 + + Returns: + 包含生成结果的状态字典 + """ + import asyncio + return asyncio.run( + self.generate_toc(technical_criteria, generation_mode, template_file) + ) \ No newline at end of file diff --git a/src/bidmaster/agents/interaction.py b/src/bidmaster/agents/interaction.py new file mode 100644 index 0000000..52362e1 --- /dev/null +++ b/src/bidmaster/agents/interaction.py @@ -0,0 +1,326 @@ +"""交互管理模块 + +提供统一的用户交互处理机制，支持多种交互模式。 +""" + +import json +import logging +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union + +import click +from rich.console import Console + +logger = logging.getLogger(__name__) + + +class InteractionMode: + """交互模式常量""" + INTERACTIVE = "interactive" # CLI交互模式 + SILENT = "silent" # 静默模式（使用默认值） + PROGRAMMATIC = "programmatic" # 程序化模式（使用预设值） + + +class InteractionHandler: + """统一的交互处理器 + + 根据不同模式处理用户交互需求。 + """ + + def __init__(self, + mode: str = InteractionMode.INTERACTIVE, + presets: Optional[Dict[str, Any]] = None): + """初始化交互处理器 + + Args: + mode: 交互模式 + presets: 预设值字典（programmatic模式使用） + """ + self.mode = mode + self.presets = presets or {} + self.console = Console() + self._interaction_history: List[Dict[str, Any]] = [] + + def __call__(self, interaction_type: str, **kwargs) -> Any: + """处理交互请求 + + Args: + interaction_type: 交互类型 ('choice', 'text', 'file_path', 'confirm') + **kwargs: 交互参数 + + Returns: + 用户输入或默认值 + """ + # 记录交互历史 + interaction_record = { + "type": interaction_type, + "params": kwargs.copy(), + "mode": self.mode + } + + try: + if self.mode == InteractionMode.SILENT: + result = kwargs.get('default') + elif self.mode == InteractionMode.PROGRAMMATIC: + result = self._handle_programmatic(**kwargs) + else: # INTERACTIVE + result = self._handle_interactive(interaction_type, **kwargs) + + interaction_record["result"] = result + return result + + except Exception as e: + logger.error(f"交互处理失败: {e}") + result = kwargs.get('default') + interaction_record["result"] = result + interaction_record["error"] = str(e) + return result + + finally: + self._interaction_history.append(interaction_record) + + def _handle_programmatic(self, **kwargs) -> Any: + """处理程序化模式""" + key = kwargs.get('key') + if key and key in self.presets: + return self.presets[key] + + # 如果没有预设值，使用默认值 + return kwargs.get('default') + + def _handle_interactive(self, interaction_type: str, **kwargs) -> Any: + """处理交互模式""" + if interaction_type == "choice": + return self._handle_choice(**kwargs) + elif interaction_type == "text": + return self._handle_text(**kwargs) + elif interaction_type == "file_path": + return self._handle_file_path(**kwargs) + elif interaction_type == "confirm": + return self._handle_confirm(**kwargs) + else: + logger.warning(f"未知交互类型: {interaction_type}") + return kwargs.get('default') + + def _handle_choice(self, + prompt: str, + options: List[Union[str, Tuple[str, str]]], + default: Optional[str] = None, + **kwargs) -> str: + """处理选择交互 + + Args: + prompt: 提示信息 + options: 选项列表，可以是字符串列表或(value, label)元组列表 + default: 默认值 + + Returns: + 选择的值 + """ + self.console.print(f"\n📝 {prompt}:", style="blue") + + # 标准化选项格式 + normalized_options = [] + for i, option in enumerate(options): + if isinstance(option, tuple): + value, label = option + normalized_options.append((value, label)) + self.console.print(f"{i + 1}. {label}") + else: + normalized_options.append((option, option)) + self.console.print(f"{i + 1}. {option}") + + # 构建选择列表 + choices = [str(i) for i in range(1, len(normalized_options) + 1)] + + try: + choice_str = click.prompt( + "请输入选择", + type=click.Choice(choices), + show_choices=False + ) + choice_index = int(choice_str) - 1 + return normalized_options[choice_index][0] + + except (click.Abort, KeyboardInterrupt): + if default is not None: + self.console.print(f"使用默认值: {default}", style="yellow") + return default + raise + + def _handle_text(self, + prompt: str, + default: Optional[str] = None, + validation: Optional[Dict[str, Any]] = None, + **kwargs) -> str: + """处理文本输入交互 + + Args: + prompt: 提示信息 + default: 默认值 + validation: 验证规则 + + Returns: + 输入的文本 + """ + while True: + try: + result = click.prompt(prompt, default=default, type=str) + + # 验证输入 + if validation: + error = self._validate_text(result, validation) + if error: + self.console.print(f"❌ {error}", style="red") + continue + + return result + + except (click.Abort, KeyboardInterrupt): + if default is not None: + return default + raise + + def _handle_file_path(self, + prompt: str, + default: Optional[str] = None, + validation: Optional[Dict[str, Any]] = None, + **kwargs) -> str: + """处理文件路径输入交互 + + Args: + prompt: 提示信息 + default: 默认值 + validation: 验证规则 + - exists: 文件必须存在 + - extensions: 允许的文件扩展名列表 + + Returns: + 文件路径 + """ + while True: + try: + path_str = click.prompt(prompt, default=default, type=str) + path = Path(path_str) + + # 验证文件路径 + if validation: + error = self._validate_file_path(path, validation) + if error: + self.console.print(f"❌ {error}", style="red") + continue + + return str(path.absolute()) + + except (click.Abort, KeyboardInterrupt): + if default is not None: + return default + raise + + def _handle_confirm(self, + prompt: str, + default: Optional[bool] = None, + **kwargs) -> bool: + """处理确认交互 + + Args: + prompt: 提示信息 + default: 默认值 + + Returns: + 确认结果 + """ + try: + return click.confirm(prompt, default=default) + except (click.Abort, KeyboardInterrupt): + if default is not None: + return default + raise + + def _validate_text(self, text: str, validation: Dict[str, Any]) -> Optional[str]: + """验证文本输入 + + Args: + text: 输入文本 + validation: 验证规则 + + Returns: + 错误信息，无错误返回None + """ + if "min_length" in validation: + if len(text) < validation["min_length"]: + return f"文本长度不能少于{validation['min_length']}个字符" + + if "max_length" in validation: + if len(text) > validation["max_length"]: + return f"文本长度不能超过{validation['max_length']}个字符" + + if "pattern" in validation: + import re + if not re.match(validation["pattern"], text): + return f"文本格式不符合要求" + + return None + + def _validate_file_path(self, path: Path, validation: Dict[str, Any]) -> Optional[str]: + """验证文件路径 + + Args: + path: 文件路径 + validation: 验证规则 + + Returns: + 错误信息，无错误返回None + """ + if validation.get("exists", False): + if not path.exists(): + return f"文件不存在: {path}" + + if "extensions" in validation: + allowed_extensions = validation["extensions"] + if path.suffix.lower() not in [ext.lower() for ext in allowed_extensions]: + return f"文件格式不支持，支持的格式: {', '.join(allowed_extensions)}" + + return None + + def get_interaction_history(self) -> List[Dict[str, Any]]: + """获取交互历史 + + Returns: + 交互历史记录列表 + """ + return self._interaction_history.copy() + + def clear_history(self) -> None: + """清空交互历史""" + self._interaction_history.clear() + + def export_presets(self, file_path: Union[str, Path]) -> None: + """导出当前会话的交互结果为预设文件 + + Args: + file_path: 导出文件路径 + """ + presets = {} + for record in self._interaction_history: + key = record["params"].get("key") + if key and "result" in record: + presets[key] = record["result"] + + with open(file_path, 'w', encoding='utf-8') as f: + json.dump(presets, f, ensure_ascii=False, indent=2) + + logger.info(f"预设配置已导出到: {file_path}") + + @classmethod + def load_presets(cls, file_path: Union[str, Path]) -> Dict[str, Any]: + """加载预设配置文件 + + Args: + file_path: 预设文件路径 + + Returns: + 预设配置字典 + """ + with open(file_path, 'r', encoding='utf-8') as f: + return json.load(f) \ No newline at end of file diff --git a/src/bidmaster/nodes/__init__.py b/src/bidmaster/nodes/__init__.py new file mode 100644 index 0000000..cda7955 --- /dev/null +++ b/src/bidmaster/nodes/__init__.py @@ -0,0 +1,8 @@ +"""节点模块 + +所有独立的业务逻辑节点，支持Agent流程组装。 +""" + +from .base import BaseNode, NodeContext + +__all__ = ["BaseNode", "NodeContext"] \ No newline at end of file diff --git a/src/bidmaster/nodes/base.py b/src/bidmaster/nodes/base.py new file mode 100644 index 0000000..cd6aa44 --- /dev/null +++ b/src/bidmaster/nodes/base.py @@ -0,0 +1,136 @@ +"""节点基础接口 + +定义所有节点的统一接口和上下文管理。 +""" + +import logging +from typing import Any, Optional, Callable, Dict +from abc import ABC, abstractmethod + +logger = logging.getLogger(__name__) + + +class NodeContext: + """节点执行上下文 + + 用于管理节点执行时的环境信息和交互处理。 + """ + + def __init__(self, interaction_handler: Optional[Callable] = None): + """初始化上下文 + + Args: + interaction_handler: 交互处理函数，签名为 (interaction_type: str, **kwargs) -> Any + """ + self.interaction_handler = interaction_handler + self.metadata: Dict[str, Any] = {} + + def interact(self, interaction_type: str, **kwargs) -> Any: + """触发交互 + + Args: + interaction_type: 交互类型，如 'choice', 'text', 'file_path' + **kwargs: 交互参数 + - prompt: 提示信息 + - options: 选项列表 (对于choice类型) + - default: 默认值 + - validation: 验证规则 + + Returns: + 用户输入或默认值 + """ + if self.interaction_handler: + try: + return self.interaction_handler(interaction_type, **kwargs) + except Exception as e: + logger.error(f"交互处理失败: {e}") + return kwargs.get('default') + return kwargs.get('default') + + def set_metadata(self, key: str, value: Any) -> None: + """设置元数据""" + self.metadata[key] = value + + def get_metadata(self, key: str, default: Any = None) -> Any: + """获取元数据""" + return self.metadata.get(key, default) + + +class BaseNode(ABC): + """节点基类 + + 所有业务逻辑节点的基础接口。 + """ + + @property + @abstractmethod + def name(self) -> str: + """节点名称（唯一标识）""" + pass + + @property + @abstractmethod + def description(self) -> str: + """节点描述""" + pass + + @abstractmethod + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行节点逻辑 + + Args: + state: 当前状态字典 + context: 节点执行上下文 + + Returns: + 更新后的状态字典 + + Raises: + Exception: 节点执行失败时抛出异常 + """ + pass + + def _log_start(self) -> None: + """记录节点开始执行""" + logger.info(f"开始执行节点: {self.name} - {self.description}") + + def _log_success(self) -> None: + """记录节点执行成功""" + logger.info(f"节点执行成功: {self.name}") + + def _log_error(self, error: Exception) -> None: + """记录节点执行失败""" + logger.error(f"节点执行失败: {self.name} - {error}") + + +class ConditionalNode(BaseNode): + """条件节点基类 + + 支持条件执行的节点。 + """ + + @abstractmethod + def should_execute(self, state: Dict[str, Any], context: NodeContext) -> bool: + """判断是否应该执行此节点 + + Args: + state: 当前状态 + context: 执行上下文 + + Returns: + 是否执行 + """ + pass + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行带条件检查的节点""" + if not self.should_execute(state, context): + logger.info(f"跳过节点执行: {self.name}") + return state + + return self._execute_impl(state, context) + + @abstractmethod + def _execute_impl(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """实际执行逻辑""" + pass \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/__init__.py b/src/bidmaster/nodes/toc/__init__.py new file mode 100644 index 0000000..2972644 --- /dev/null +++ b/src/bidmaster/nodes/toc/__init__.py @@ -0,0 +1,18 @@ +"""目录生成相关节点 + +包含目录生成Agent的所有节点实现。 +""" + +from .group_criteria import GroupCriteriaNode +from .generate_first_level import GenerateFirstLevelNode +from .generate_sub_chapters import GenerateSubChaptersNode +from .review_structure import ReviewStructureNode +from .finalize_chapters import FinalizeChaptersNode + +__all__ = [ + "GroupCriteriaNode", + "GenerateFirstLevelNode", + "GenerateSubChaptersNode", + "ReviewStructureNode", + "FinalizeChaptersNode" +] \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/finalize_chapters.py b/src/bidmaster/nodes/toc/finalize_chapters.py new file mode 100644 index 0000000..b2cb126 --- /dev/null +++ b/src/bidmaster/nodes/toc/finalize_chapters.py @@ -0,0 +1,141 @@ +"""最终确定章节结构的节点 + +基于AI审查结果最终确定章节结构，并确保包含必要的标准章节。 +""" + +import logging +from typing import Dict, List, Any + +from ..base import BaseNode, NodeContext +from ...tools.parser import DocumentChapter + +logger = logging.getLogger(__name__) + + +class FinalizeChaptersNode(BaseNode): + """最终确定章节结构的节点""" + + @property + def name(self) -> str: + return "finalize_chapters" + + @property + def description(self) -> str: + return "最终确定章节结构" + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行章节最终确定""" + self._log_start() + + try: + preliminary_chapters = state.get("preliminary_chapters", []) + structure_review = state.get("structure_review", {}) + + if not preliminary_chapters: + raise ValueError("缺少初步章节数据") + + # 应用审查建议并最终确定 + final_chapters = self._finalize_with_review(preliminary_chapters, structure_review) + + # 确保标准章节存在 + final_chapters = self._ensure_standard_chapters(final_chapters) + + logger.info(f"最终生成{len(final_chapters)}个章节") + + # 更新状态 + state["final_chapters"] = final_chapters + state["current_step"] = self.name + state["should_continue"] = False # 完成流程 + + self._log_success() + return state + + except Exception as e: + self._log_error(e) + state["error"] = str(e) + state["should_continue"] = False + raise + + def _finalize_with_review(self, + preliminary_chapters: List[DocumentChapter], + structure_review: Dict[str, Any]) -> List[DocumentChapter]: + """根据审查结果最终确定章节 + + Args: + preliminary_chapters: 初步章节列表 + structure_review: AI审查结果 + + Returns: + 最终章节列表 + """ + final_chapters = preliminary_chapters.copy() + + # 应用高优先级建议 + suggestions = structure_review.get("suggestions", []) + high_priority_suggestions = [ + s for s in suggestions + if s.get("priority") == "high" + ] + + for suggestion in high_priority_suggestions: + suggestion_type = suggestion.get("type", "") + description = suggestion.get("description", "") + + if suggestion_type == "add": + logger.info(f"应用高优先级建议-添加: {description}") + # 这里可以根据具体建议内容添加章节 + # 当前版本保持简单，只记录日志 + + elif suggestion_type == "modify": + logger.info(f"应用高优先级建议-修改: {description}") + # 这里可以根据具体建议内容修改章节 + # 当前版本保持简单，只记录日志 + + elif suggestion_type == "reorder": + logger.info(f"应用高优先级建议-重排: {description}") + # 这里可以根据具体建议内容重新排序 + # 当前版本保持简单，只记录日志 + + return final_chapters + + def _ensure_standard_chapters(self, chapters: List[DocumentChapter]) -> List[DocumentChapter]: + """确保包含标准章节 + + Args: + chapters: 当前章节列表 + + Returns: + 包含标准章节的完整列表 + """ + final_chapters = chapters.copy() + + # 确保评标索引表存在（作为第一章） + has_index = any("评标索引" in ch.title for ch in final_chapters) + if not has_index: + index_chapter = DocumentChapter( + id="evaluation_index", + title="评标索引表（技术评分完全对应）", # 不包含编号 + level=1, + template_placeholder="{{evaluation_index_content}}" + ) + final_chapters.insert(0, index_chapter) + logger.info("添加评标索引表章节") + + # 可以在此添加其他必要的标准章节 + # 例如：封面、目录、声明等 + + return final_chapters + + def _log_chapter_structure(self, chapters: List[DocumentChapter]) -> None: + """记录最终章节结构 + + Args: + chapters: 章节列表 + """ + logger.info("最终章节结构:") + for chapter in chapters: + logger.info(f" {chapter.title}") + for sub in chapter.children: + logger.info(f" {sub.title}") + for child in sub.children: + logger.info(f" {child.title}") \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/generate_first_level.py b/src/bidmaster/nodes/toc/generate_first_level.py new file mode 100644 index 0000000..93962ca --- /dev/null +++ b/src/bidmaster/nodes/toc/generate_first_level.py @@ -0,0 +1,123 @@ +"""生成一级章节的节点 + +根据评分项类别生成一级章节结构。 +""" + +import logging +from typing import Dict, List, Any + +from ..base import BaseNode, NodeContext +from ...tools.parser import ScoringCriteria, DocumentChapter + +logger = logging.getLogger(__name__) + +# 统一的类别名称映射 +CATEGORY_NAMES = { + "compliance": "合规响应", + "technical_solution": "技术方案", + "equipment_spec": "设备规格", + "quality_safety": "质量安全", + "after_sales": "售后服务", + "implementation": "实施方案" +} + + +class GenerateFirstLevelNode(BaseNode): + """生成一级章节的节点""" + + @property + def name(self) -> str: + return "generate_first_level" + + @property + def description(self) -> str: + return "生成一级章节" + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行一级章节生成""" + self._log_start() + + try: + category_groups = state.get("category_groups", {}) + technical_criteria = state.get("technical_criteria", []) + + if not category_groups: + raise ValueError("缺少类别分组数据") + + # 生成一级章节 + chapters = self._generate_first_level_chapters(category_groups, technical_criteria) + + logger.info(f"生成{len(chapters)}个一级章节") + + # 更新状态 + state["preliminary_chapters"] = chapters + state["current_step"] = self.name + + self._log_success() + return state + + except Exception as e: + self._log_error(e) + state["error"] = str(e) + state["should_continue"] = False + raise + + def _generate_first_level_chapters(self, + category_groups: Dict[str, List[ScoringCriteria]], + technical_criteria: List[ScoringCriteria]) -> List[DocumentChapter]: + """生成一级章节 + + Args: + category_groups: 按类别分组的评分项 + technical_criteria: 原始技术评分项列表（用于确定顺序） + + Returns: + 一级章节列表 + """ + chapters = [] + chapter_index = 1 + + # 按原始顺序确定章节顺序 + def get_category_first_index(category: str) -> int: + """获取类别在原始评分项中的首次出现位置""" + for i, criteria in enumerate(technical_criteria): + if criteria.category.value == category: + return i + return 999 # 未找到时排在最后 + + # 按原始出现顺序排序类别 + sorted_categories = sorted(category_groups.keys(), key=get_category_first_index) + + # 为每个类别创建一级章节 + for category in sorted_categories: + criteria_list = category_groups[category] + if not criteria_list: + continue + + # 计算类别总分 + total_score = sum(c.max_score for c in criteria_list) + + # 生成章节ID和标题 + chapter_id = f"chapter_{chapter_index:02d}_{category}" + category_name = CATEGORY_NAMES.get(category, category) + chapter_title = category_name + + # 添加分值信息（如果有分值） + if total_score > 0: + chapter_title += f" ({total_score}分)" + + # 创建章节对象 + chapter = DocumentChapter( + id=chapter_id, + title=chapter_title, + level=1, + score=total_score, + template_placeholder=f"{{{{{chapter_id}_content}}}}" + ) + + chapters.append(chapter) + chapter_index += 1 + + logger.debug(f"创建一级章节: {chapter_title}") + + return chapters \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/generate_sub_chapters.py b/src/bidmaster/nodes/toc/generate_sub_chapters.py new file mode 100644 index 0000000..15de377 --- /dev/null +++ b/src/bidmaster/nodes/toc/generate_sub_chapters.py @@ -0,0 +1,294 @@ +"""生成二三级子标题的节点 + +通过AI智能生成或模板生成二三级子标题。 +""" + +import json +import logging +from typing import Dict, List, Any, Optional + +from ..base import BaseNode, NodeContext +from ...tools.parser import ScoringCriteria, DocumentChapter +from ...tools.llm import llm_service + +logger = logging.getLogger(__name__) + + +class GenerateSubChaptersNode(BaseNode): + """生成二三级子标题的节点""" + + @property + def name(self) -> str: + return "generate_sub_chapters" + + @property + def description(self) -> str: + return "生成二三级子标题" + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行子标题生成""" + self._log_start() + + try: + preliminary_chapters = state.get("preliminary_chapters", []) + technical_criteria = state.get("technical_criteria", []) + + if not preliminary_chapters: + raise ValueError("缺少初步章节") + + # 通过context处理交互 + generation_mode = context.interact( + "choice", + key="generation_mode", + prompt="请选择章节内容生成方式", + options=[ + ("ai", "AI智能生成子标题（推荐）- 根据评分项要求智能生成2-3级标题"), + ("template", "基于模板生成 - 使用预定义模板结构") + ], + default="ai" + ) + + template_file = None + if generation_mode == "template": + template_file = context.interact( + "file_path", + key="template_file", + prompt="请输入模板文件路径（.docx格式）", + validation={ + "exists": True, + "extensions": [".docx"] + } + ) + + # 为每个章节生成子标题 + enhanced_chapters = [] + for chapter in preliminary_chapters: + enhanced_chapter = self._enhance_chapter_with_subs( + chapter, technical_criteria, generation_mode, template_file + ) + enhanced_chapters.append(enhanced_chapter) + + logger.info(f"完成{len(enhanced_chapters)}个章节的子标题生成") + + # 更新状态 + state["preliminary_chapters"] = enhanced_chapters + state["current_step"] = self.name + + self._log_success() + return state + + except Exception as e: + self._log_error(e) + state["error"] = str(e) + state["should_continue"] = False + raise + + def _enhance_chapter_with_subs(self, + chapter: DocumentChapter, + technical_criteria: List[ScoringCriteria], + generation_mode: str, + template_file: Optional[str]) -> DocumentChapter: + """为章节增强子标题 + + Args: + chapter: 原始章节 + technical_criteria: 技术评分项 + generation_mode: 生成模式 + template_file: 模板文件（仅template模式使用） + + Returns: + 增强后的章节 + """ + # 找到该章节对应的评分项 + corresponding_criteria = self._find_corresponding_criteria(chapter, technical_criteria) + + if not corresponding_criteria: + logger.warning(f"章节 {chapter.title} 没有找到对应的评分项") + return chapter + + # 根据模式生成子标题 + if generation_mode == "ai": + sub_chapters = self._generate_ai_sub_chapters(corresponding_criteria, chapter) + else: + sub_chapters = self._generate_template_sub_chapters( + corresponding_criteria[0], chapter, template_file + ) + + # 设置子章节 + chapter.children = sub_chapters + logger.info(f"章节 {chapter.title} 生成了 {len(sub_chapters)} 个子标题") + + return chapter + + def _find_corresponding_criteria(self, + chapter: DocumentChapter, + technical_criteria: List[ScoringCriteria]) -> List[ScoringCriteria]: + """查找章节对应的评分项 + + Args: + chapter: 章节 + technical_criteria: 技术评分项列表 + + Returns: + 对应的评分项列表 + """ + # 从章节ID提取类别 + if "_" not in chapter.id: + return [] + + parts = chapter.id.split("_") + if len(parts) < 3: + return [] + + category = "_".join(parts[2:]) + + # 找到该类别的所有评分项 + return [ + c for c in technical_criteria + if c.category.value == category + ] + + def _generate_ai_sub_chapters(self, + criteria_list: List[ScoringCriteria], + parent_chapter: DocumentChapter) -> List[DocumentChapter]: + """AI生成子标题 + + Args: + criteria_list: 对应的评分项列表 + parent_chapter: 父章节 + + Returns: + 子章节列表 + """ + try: + # 构建评分项信息 + criteria_info = [] + for criteria in criteria_list: + criteria_info.append(f"- {criteria.item_name} ({criteria.max_score}分)") + + prompt = f""" +为以下大类别生成专业的标书子标题： + +【大类别】: {parent_chapter.title} +【评分项】: +{chr(10).join(criteria_info)} + +生成要求： +1. 为每个评分项生成对应的子标题名称（不要包含编号） +2. 重要评分项可添加三级子标题（不要包含编号） +3. 只返回标题文本，编号由Word自动管理 + +返回JSON格式： +{{ + "sub_chapters": [ + {{"title": "技术架构设计", "level": 2, "score": 5, "children": []}} + ] +}} + +只返回JSON：""" + + # 使用统一的LLM服务 + response = llm_service.call(prompt) + if not response: + raise ValueError("AI生成子标题失败: API无响应") + + # 解析响应 + result_data = self._parse_ai_response(response) + sub_chapters_data = result_data.get("sub_chapters", []) + + # 构建子章节对象 + sub_chapters = [] + for i, sub_data in enumerate(sub_chapters_data, 1): + title = sub_data.get("title", f"子标题{i}") + + sub_chapter = DocumentChapter( + id=f"{parent_chapter.id}_sub_{i:02d}", + title=title, # 不添加编号 + level=sub_data.get("level", 2), + score=sub_data.get("score", 0), + template_placeholder=f"{{{{{parent_chapter.id}_sub_{i:02d}_content}}}}" + ) + + # 处理三级标题 + for j, child_data in enumerate(sub_data.get("children", []), 1): + child_title = child_data.get("title", f"三级标题{j}") + + child_chapter = DocumentChapter( + id=f"{parent_chapter.id}_sub_{i:02d}_{j:02d}", + title=child_title, # 不添加编号 + level=child_data.get("level", 3), + template_placeholder=f"{{{{{parent_chapter.id}_sub_{i:02d}_{j:02d}_content}}}}" + ) + sub_chapter.children.append(child_chapter) + + sub_chapters.append(sub_chapter) + + return sub_chapters + + except Exception as e: + logger.error(f"AI生成子标题失败: {e}") + # 立即失败，不提供后备方案 + raise + + def _parse_ai_response(self, response: str) -> Dict[str, Any]: + """解析AI响应 + + Args: + response: AI响应文本 + + Returns: + 解析后的数据 + + Raises: + ValueError: 解析失败时抛出 + """ + try: + clean_response = response.strip() + if clean_response.startswith("```json"): + clean_response = clean_response[7:] + if clean_response.endswith("```"): + clean_response = clean_response[:-3] + + return json.loads(clean_response.strip()) + + except (json.JSONDecodeError, KeyError) as e: + raise ValueError(f"解析AI响应失败: {e}") + + def _generate_template_sub_chapters(self, + criteria: ScoringCriteria, + parent_chapter: DocumentChapter, + template_file: Optional[str]) -> List[DocumentChapter]: + """基于模板生成子标题 + + Args: + criteria: 评分项（仅作参考） + parent_chapter: 父章节 + template_file: 模板文件路径 + + Returns: + 子章节列表 + """ + # 提供默认结构（不包含编号） + default_sub_chapters = [ + DocumentChapter( + id=f"{parent_chapter.id}_def_01", + title="方案概述", # 不包含编号 + level=2, + template_placeholder=f"{{{{{parent_chapter.id}_def_01_content}}}}" + ), + DocumentChapter( + id=f"{parent_chapter.id}_def_02", + title="具体实施", # 不包含编号 + level=2, + template_placeholder=f"{{{{{parent_chapter.id}_def_02_content}}}}" + ), + DocumentChapter( + id=f"{parent_chapter.id}_def_03", + title="保障措施", # 不包含编号 + level=2, + template_placeholder=f"{{{{{parent_chapter.id}_def_03_content}}}}" + ) + ] + + logger.info(f"使用默认模板为 {parent_chapter.title} 生成子标题") + return default_sub_chapters \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/group_criteria.py b/src/bidmaster/nodes/toc/group_criteria.py new file mode 100644 index 0000000..8017495 --- /dev/null +++ b/src/bidmaster/nodes/toc/group_criteria.py @@ -0,0 +1,91 @@ +"""按类别对评分项分组的节点 + +将技术评分项按照类别进行分组，为后续章节生成做准备。 +""" + +import logging +from typing import Dict, List, Any + +from ..base import BaseNode, NodeContext +from ...tools.parser import ScoringCriteria + +logger = logging.getLogger(__name__) + + +class GroupCriteriaNode(BaseNode): + """按类别对评分项分组的节点""" + + @property + def name(self) -> str: + return "group_criteria" + + @property + def description(self) -> str: + return "按类别对评分项分组" + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行分组逻辑""" + self._log_start() + + try: + technical_criteria = state.get("technical_criteria", []) + + if not technical_criteria: + raise ValueError("缺少技术评分项") + + # 执行分组 + category_groups = self._group_by_category(technical_criteria) + + logger.info(f"分组完成: {len(category_groups)}个类别") + + # 更新状态 + state["category_groups"] = category_groups + state["current_step"] = self.name + + self._log_success() + return state + + except Exception as e: + self._log_error(e) + state["error"] = str(e) + state["should_continue"] = False + raise + + def _group_by_category(self, criteria: List[ScoringCriteria]) -> Dict[str, List[ScoringCriteria]]: + """按类别分组评分项 + + Args: + criteria: 评分项列表 + + Returns: + 分组后的字典，key为类别，value为该类别下的评分项列表 + """ + # 预定义类别顺序 + category_groups = { + "technical_solution": [], + "equipment_spec": [], + "implementation": [], + "quality_safety": [], + "after_sales": [], + "compliance": [] + } + + # 分组 + for criterion in criteria: + category_key = criterion.category.value + if category_key in category_groups: + category_groups[category_key].append(criterion) + else: + # 未知类别默认归入技术方案 + category_groups["technical_solution"].append(criterion) + logger.warning(f"未知类别 {category_key}，归入技术方案类别") + + # 只保留有评分项的类别 + filtered_groups = {k: v for k, v in category_groups.items() if v} + + # 记录分组统计 + for category, items in filtered_groups.items(): + total_score = sum(item.max_score for item in items) + logger.info(f"类别 {category}: {len(items)}项，总分 {total_score}") + + return filtered_groups \ No newline at end of file diff --git a/src/bidmaster/nodes/toc/review_structure.py b/src/bidmaster/nodes/toc/review_structure.py new file mode 100644 index 0000000..eaba659 --- /dev/null +++ b/src/bidmaster/nodes/toc/review_structure.py @@ -0,0 +1,200 @@ +"""AI审查目录结构的节点 + +使用AI对生成的目录结构进行合理性和完整性审查。 +""" + +import json +import logging +from typing import Dict, List, Any + +from ..base import BaseNode, NodeContext +from ...tools.parser import ScoringCriteria, DocumentChapter +from ...tools.llm import llm_service + +logger = logging.getLogger(__name__) + +# 类别名称映射（用于审查显示） +CATEGORY_NAMES = { + "compliance": "合规响应", + "technical_solution": "技术方案", + "equipment_spec": "设备规格", + "quality_safety": "质量安全", + "after_sales": "售后服务", + "implementation": "实施方案" +} + + +class ReviewStructureNode(BaseNode): + """AI审查目录结构的节点""" + + @property + def name(self) -> str: + return "review_structure" + + @property + def description(self) -> str: + return "AI审查目录结构" + + def execute(self, state: Dict[str, Any], context: NodeContext) -> Dict[str, Any]: + """执行AI审查""" + self._log_start() + + try: + technical_criteria = state.get("technical_criteria", []) + preliminary_chapters = state.get("preliminary_chapters", []) + + if not preliminary_chapters: + raise ValueError("缺少章节数据进行审查") + + # 执行AI审查 + review_result = self._perform_ai_review(technical_criteria, preliminary_chapters) + + # 更新状态 + state["structure_review"] = review_result + state["current_step"] = self.name + + # 根据审查结果添加警告 + if review_result.get("suggestions"): + suggestion_count = len(review_result["suggestions"]) + state.setdefault("warnings", []).append(f"AI审查: {suggestion_count}条优化建议") + + logger.info(f"AI审查完成，优化评分: {review_result.get('optimization_score', 'N/A')}") + self._log_success() + return state + + except Exception as e: + self._log_error(e) + # AI审查失败不中断流程，只记录警告 + logger.warning(f"AI审查失败: {e}") + state.setdefault("warnings", []).append(f"AI审查跳过: {str(e)}") + state["structure_review"] = {} + state["current_step"] = self.name + return state + + def _perform_ai_review(self, + technical_criteria: List[ScoringCriteria], + preliminary_chapters: List[DocumentChapter]) -> Dict[str, Any]: + """执行AI审查 + + Args: + technical_criteria: 技术评分项 + preliminary_chapters: 初步生成的章节 + + Returns: + 审查结果字典 + """ + # 构建审查提示词 + criteria_summary = self._format_criteria_for_review(technical_criteria) + chapters_summary = self._format_chapters_for_review(preliminary_chapters) + + review_prompt = f""" +请审查这个标书目录结构的合理性和完整性。 + +【技术评分项分布】: +{criteria_summary} + +【当前生成的章节结构】: +{chapters_summary} + +【审查要求】: +1. 是否缺少重要的标准章节？ +2. 章节顺序是否合理？ +3. 每个评分项是否都有对应章节？ + +返回JSON格式： +{{ + "overall_assessment": "总体评价", + "suggestions": [ + {{"type": "add/modify/reorder", "description": "建议内容", "priority": "high/medium/low"}} + ], + "optimization_score": 85 +}} + +只返回JSON：""" + + try: + # 使用统一的LLM服务 + response = llm_service.call(review_prompt) + + if response: + return self._parse_review_response(response) + else: + return {"overall_assessment": "AI审查失败", "suggestions": []} + + except Exception as e: + logger.error(f"AI审查API调用失败: {e}") + return {"overall_assessment": "AI审查异常", "suggestions": []} + + def _parse_review_response(self, response: str) -> Dict[str, Any]: + """解析AI审查响应 + + Args: + response: AI响应文本 + + Returns: + 解析后的审查结果 + """ + try: + clean_response = response.strip() + if clean_response.startswith("```json"): + clean_response = clean_response[7:] + if clean_response.endswith("```"): + clean_response = clean_response[:-3] + + return json.loads(clean_response.strip()) + + except json.JSONDecodeError as e: + logger.error(f"解析AI审查响应失败: {e}") + return {"overall_assessment": "解析失败", "suggestions": []} + + def _format_criteria_for_review(self, technical_criteria: List[ScoringCriteria]) -> str: + """格式化评分项用于审查 + + Args: + technical_criteria: 技术评分项列表 + + Returns: + 格式化后的字符串 + """ + lines = [] + + # 按类别分组 + category_groups = {} + for criteria in technical_criteria: + category = criteria.category.value + if category not in category_groups: + category_groups[category] = [] + category_groups[category].append(criteria) + + # 格式化输出 + for category, items in category_groups.items(): + category_name = CATEGORY_NAMES.get(category, category) + lines.append(f"【{category_name}】({len(items)}项):") + for item in items: + lines.append(f" - {item.item_name} ({item.max_score}分)") + + return "\n".join(lines) + + def _format_chapters_for_review(self, chapters: List[DocumentChapter]) -> str: + """格式化章节用于审查 + + Args: + chapters: 章节列表 + + Returns: + 格式化后的字符串 + """ + lines = [] + + for chapter in chapters: + lines.append(chapter.title) + + # 显示子章节 + for sub in chapter.children: + lines.append(f" {sub.title}") + + # 显示三级章节 + for child in sub.children: + lines.append(f" {child.title}") + + return "\n".join(lines) \ No newline at end of file