(
record_attributes: list[str],
chunk_key: str,
)
| 23 | results = re.split("|".join(re.escape(marker) for marker in markers), content) |
| 24 | return [r.strip() for r in results if r.strip()] |
| 25 | async def _handle_single_entity_extraction( |
| 26 | record_attributes: list[str], |
| 27 | chunk_key: str, |
| 28 | ): |
| 29 | if len(record_attributes) < 4 or record_attributes[0] != '"entity"': |
| 30 | return None |
| 31 | # add this record as a node in the G |
| 32 | entity_name = clean_str(record_attributes[1].upper()) |
| 33 | if not entity_name.strip(): |
| 34 | return None |
| 35 | entity_type = clean_str(record_attributes[2].upper()) |
| 36 | entity_description = clean_str(record_attributes[3]) |
| 37 | entity_source_id = chunk_key |
| 38 | return dict( |
| 39 | entity_name=entity_name, |
| 40 | entity_type=entity_type, |
| 41 | description=entity_description, |
| 42 | source_id=entity_source_id, |
| 43 | ) |
| 44 | def is_float_regex(value): |
| 45 | return bool(re.match(r"^[-+]?[0-9]*\.?[0-9]+$", value)) |
| 46 | async def _handle_single_relationship_extraction( |
no test coverage detected