video_decode/src/components

组件抽象层设计说明

概述

本组件抽象层为 AIGC 解构系统提供了统一的原子组件接口，支持 Agent、Tool 和 Function 三种类型的组件。所有组件都遵循单一职责原则，可以灵活组合构建复杂的工作流。

架构设计

组件抽象层
├── agents/           # Agent 组件
│   └── base.py      # Agent 基类定义
├── tools/           # Tool 组件  
│   └── base.py      # Tool 基类定义
├── functions/       # Function 组件
│   └── base.py      # Function 基类定义
└── __init__.py      # 统一导入接口

组件类型

1. Agent 组件

Agent 组件负责智能决策和复杂推理，基于 LangGraph 实现。

基类层次

BaseComponent (抽象基类)
    └── BaseAgent (Agent 基类)
        ├── BaseLLMAgent (LLM Agent)
        ├── BaseReactAgent (ReAct Agent) 
        └── BaseGraphAgent (Graph Agent)

Agent 类型说明

• BaseLLMAgent: 直接基于 LLM 的简单智能体，适用于单轮对话和简单任务
• BaseReactAgent: 基于 ReAct 模式的智能体，支持工具调用和推理
• BaseGraphAgent: 基于 StateGraph 的复杂智能体，支持多节点工作流

2. Tool 组件

Tool 组件提供外部能力集成，支持 LangGraph 的 @tool 装饰器。

Tool 类型

• SimpleTool: 基于函数的简单工具
• AsyncTool: 支持异步操作的工具
• ConfigurableTool: 支持运行时配置的工具

工具注册表

from src.components import tool_registry

# 获取所有启用的工具
enabled_tools = tool_registry.get_enabled_tools()

3. Function 组件

Function 组件处理纯函数计算，无副作用的数据转换操作。

Function 类型

• SimpleFunction: 基于普通函数的简单包装
• ConfigurableFunction: 支持配置参数的函数组件
• PipelineFunction: 将多个函数组件串联成管道
• ConditionalFunction: 根据条件选择不同的函数执行

使用示例

创建 Agent

from src.components import BaseLLMAgent, BaseReactAgent

# 创建 LLM Agent
class PersonaAnalysisAgent(BaseLLMAgent):
    def __init__(self):
        super().__init__(
            name="persona_analysis",
            description="分析用户人设特征",
            system_prompt="你是专业的人设分析师..."
        )
    
    def _build_messages(self, state):
        # 实现消息构建逻辑
        pass
    
    def _update_state(self, state, response):
        # 实现状态更新逻辑
        pass

# 创建 ReAct Agent
class DeconstructionAgent(BaseReactAgent):
    def __init__(self, tools):
        super().__init__(
            name="deconstruction",
            description="执行内容解构",
            tools=tools
        )
    
    def _extract_messages(self, state):
        # 从状态中提取消息
        pass
    
    def _merge_response(self, state, response):
        # 合并响应到状态
        pass

创建 Tool

from src.components import component_tool
from pydantic import BaseModel

class SearchInput(BaseModel):
    query: str
    limit: int = 10

@component_tool(
    name="search_tool",
    description="搜索相关信息",
    category="search",
    schema=SearchInput
)
def search_information(query: str, limit: int = 10) -> str:
    \"\"\"搜索相关信息\"\"\"
    # 实现搜索逻辑
    return "搜索结果..."

创建 Function

from src.components import component_function

@component_function(
    name="data_processor",
    description="处理数据格式",
    category="data_processing"
)
def process_data(input_data: dict, context: dict = None) -> dict:
    \"\"\"处理数据格式\"\"\"
    # 实现数据处理逻辑
    return processed_data

组合使用

from src.components import tool_registry, function_registry
from langgraph.graph import StateGraph

# 获取工具
search_tools = tool_registry.get_tools_by_category("search")

# 创建 Agent
agent = DeconstructionAgent(tools=[tool.get_tool() for tool in search_tools])

# 在工作流中使用
def agent_node(state):
    return agent.process(state)

workflow = StateGraph(state_schema)
workflow.add_node("agent", agent_node)

设计原则

1. 单一职责

每个组件只负责一个明确的功能，便于测试和维护。

2. 接口统一

所有同类型组件都实现相同的基础接口，便于替换和扩展。

3. 配置灵活

支持运行时配置，可以根据不同场景调整组件行为。

4. 注册管理

提供注册表机制，统一管理组件的生命周期。

5. LangGraph 集成

深度集成 LangGraph 框架，充分利用其状态管理和工作流能力。

扩展指南

添加新的 Agent 类型

1. 继承相应的 BaseAgent 子类
2. 实现必需的抽象方法
3. 在工作流中使用

添加新的 Tool

1. 使用 @component_tool 装饰器
2. 定义输入 schema (可选)
3. 自动注册到 tool_registry

添加新的 Function

1. 使用 @component_function 装饰器
2. 确保函数无副作用
3. 自动注册到 function_registry

最佳实践

1. 命名规范: 使用清晰、描述性的名称
2. 文档字符串: 为所有组件提供详细的文档
3. 错误处理: 实现适当的异常处理
4. 日志记录: 使用统一的日志记录机制
5. 测试覆盖: 为每个组件编写单元测试

注意事项

• Agent 必须基于 LangGraph 实现
• Tool 需要加上 LangGraph 注解 (@tool)
• Function 应该是纯函数，避免副作用
• 所有组件都支持懒初始化
• 注册表是全局单例，注意线程安全