OpenClaw 基本结构

openclaw openclaw中文博客 2026-04-09 2

我来帮你了解 OpenClaw 的手动适配流程，OpenClaw 是一个用于半结构化文档信息提取的框架，手动适配通常涉及以下步骤：

OpenClaw 基本结构-第1张图片-OpenClaw 中文版 - 真正能做事的 AI

理解 OpenClaw 架构

# ClawMachine 包含：
# - States: 状态定义
# - Transitions: 状态转移规则
# - Actions: 提取动作

手动适配的常见场景

场景1：自定义提取规则

from openclaw import State, Transition, Action
from openclaw.extractors import RegexExtractor, XPathExtractor
class CustomClawMachine(ClawMachine):
    def __init__(self):
        super().__init__()
        self._setup_states()
        self._setup_transitions()
    def _setup_states(self):
        # 定义状态
        self.add_state(State("start", is_initial=True))
        self.add_state(State("extract_title"))
        self.add_state(State("extract_price"))
        self.add_state(State("end", is_final=True))
    def _setup_transitions(self):
        # 手动定义状态转移规则
        self.add_transition(
            Transition(
                from_state="start",
                to_state="extract_title",
                condition=lambda doc: self._has_title(doc)
            )
        )
    def _has_title(self, document):
        # 自定义条件判断逻辑
        return "title" in document.text.lower()

场景2：适配特定文档格式

class PDFClawAdapter:
    """手动适配 PDF 文档"""
    def __init__(self, pdf_path):
        self.pdf_path = pdf_path
        self.claw_machine = ClawMachine()
    def adapt(self):
        # 1. PDF 文本提取
        text = self._extract_text_from_pdf()
        # 2. 结构解析
        sections = self._parse_sections(text)
        # 3. 转换为 OpenClaw 可处理格式
        return self._create_document_object(sections)
    def _extract_text_from_pdf(self):
        # 使用 PyPDF2 或 pdfplumber
        import pdfplumber
        with pdfplumber.open(self.pdf_path) as pdf:
            text = ""
            for page in pdf.pages:
                text += page.extract_text()
        return text
    def _create_document_object(self, sections):
        # 构建符合 OpenClaw 格式的文档对象
        return {
            "metadata": {"type": "pdf", "sections": len(sections)},
            "content": sections,
            "annotations": self._generate_annotations(sections)
        }

场景3：自定义提取器

from openclaw.extractors import BaseExtractor
class CustomTableExtractor(BaseExtractor):
    """手动实现表格提取器"""
    def __init__(self, config=None):
        super().__init__(config)
        self.table_patterns = config.get("patterns", [])
    def extract(self, document, context=None):
        results = []
        # 手动解析逻辑
        for pattern in self.table_patterns:
            if pattern["type"] == "regex":
                matches = self._extract_by_regex(document, pattern)
            elif pattern["type"] == "structure":
                matches = self._extract_by_structure(document, pattern)
            results.extend(matches)
        return self._post_process(results)
    def _extract_by_structure(self, document, pattern):
        """基于文档结构提取"""
        tables = []
        current_table = []
        for line in document.lines:
            if self._is_table_start(line, pattern):
                current_table = [line]
            elif self._is_table_row(line, pattern) and current_table:
                current_table.append(line)
            elif self._is_table_end(line, pattern) and current_table:
                tables.append(current_table)
                current_table = []
        return tables

完整的手动适配示例

class ManualOpenClawAdapter:
    """完整的手动适配示例"""
    def __init__(self, config):
        self.config = config
        self.claw_machine = self._build_claw_machine()
    def _build_claw_machine(self):
        """手动构建 ClawMachine"""
        machine = ClawMachine()
        # 1. 添加自定义状态
        states_config = self.config.get("states", [])
        for state_config in states_config:
            state = State(
                name=state_config["name"],
                extractors=self._build_extractors(state_config),
                is_initial=state_config.get("is_initial", False),
                is_final=state_config.get("is_final", False)
            )
            machine.add_state(state)
        # 2. 添加手动定义的转移规则
        transitions_config = self.config.get("transitions", [])
        for trans_config in transitions_config:
            transition = Transition(
                from_state=trans_config["from"],
                to_state=trans_config["to"],
                condition=self._build_condition(trans_config),
                priority=trans_config.get("priority", 0)
            )
            machine.add_transition(transition)
        return machine
    def _build_extractors(self, state_config):
        """手动构建提取器集合"""
        extractors = []
        for extractor_config in state_config.get("extractors", []):
            if extractor_config["type"] == "regex":
                extractor = RegexExtractor(
                    patterns=extractor_config["patterns"],
                    field_name=extractor_config["field"]
                )
            elif extractor_config["type"] == "custom":
                extractor = self._create_custom_extractor(extractor_config)
            extractors.append(extractor)
        return extractors
    def _build_condition(self, trans_config):
        """手动构建转移条件"""
        condition_type = trans_config.get("condition_type", "function")
        if condition_type == "function":
            # 动态创建条件函数
            code = trans_config["condition_code"]
            return eval(f"lambda doc, ctx: {code}")
        elif condition_type == "regex":
            pattern = trans_config["pattern"]
            return lambda doc, ctx: bool(re.search(pattern, doc.text))
    def process_document(self, document):
        """处理文档"""
        # 预处理文档
        processed_doc = self._preprocess(document)
        # 执行状态机
        result = self.claw_machine.run(processed_doc)
        # 后处理结果
        return self._postprocess(result)
    def _preprocess(self, document):
        """文档预处理"""
        # 清理文本
        document.text = self._clean_text(document.text)
        # 添加段落标记
        document.annotations = self._detect_paragraphs(document.text)
        return document
    def _postprocess(self, result):
        """结果后处理"""
        # 合并重复字段
        merged = self._merge_duplicates(result)
        # 验证提取结果
        validated = self._validate(merged)
        # 格式化输出
        return self._format_output(validated)

配置文件示例 (YAML)

# config.yaml
states:
  - name: "start"
    is_initial: true
    extractors:
      - type: "regex"
        field: "document_type"
        patterns: ["合同", "协议", "agreement"]
  - name: "extract_parties"
    extractors:
      - type: "custom"
        class: "PartyExtractor"
        config:
          keywords: ["甲方", "乙方", "丙方"]
  - name: "extract_terms"
    is_final: true
transitions:
  - from: "start"
    to: "extract_parties"
    condition_type: "regex"
    pattern: ".*双方.*"
  - from: "extract_parties"
    to: "extract_terms"
    condition_type: "function"
    condition_code: "len(ctx.get('parties', [])) >= 2"

调试和优化建议

class OpenClawDebugger:
    """手动适配调试工具"""
    @staticmethod
    def trace_execution(machine, document):
        """跟踪执行过程"""
        print("=== Execution Trace ===")
        current_state = machine.initial_state
        while current_state and not current_state.is_final:
            print(f"Current State: {current_state.name}")
            # 检查转移条件
            for transition in machine.get_transitions(current_state):
                if transition.condition(document, machine.context):
                    print(f"  -> Transition to: {transition.to_state.name}")
                    current_state = machine.get_state(transition.to_state)
                    break
    @staticmethod
    def validate_extraction(result, expected_fields):
        """验证提取结果"""
        missing = []
        for field in expected_fields:
            if field not in result or not result[field]:
                missing.append(field)
        if missing:
            print(f"Missing fields: {missing}")
            return False
        return True