(
record_attributes: list[str],
chunk_key: str,
)
| 301 | return results |
| 302 | |
| 303 | async def _handle_single_entity_extraction( |
| 304 | record_attributes: list[str], |
| 305 | chunk_key: str, |
| 306 | ): |
| 307 | if len(record_attributes) < 4 or record_attributes[0] != '"entity"': |
| 308 | return None |
| 309 | # add this record as a node in the G |
| 310 | entity_name = clean_str(record_attributes[1].upper()) |
| 311 | if not entity_name.strip(): |
| 312 | return None |
| 313 | entity_type = clean_str(record_attributes[2].upper()) |
| 314 | entity_description = clean_str(record_attributes[3]) |
| 315 | entity_source_id = chunk_key |
| 316 | return dict( |
| 317 | entity_name=entity_name, |
| 318 | entity_type=entity_type, |
| 319 | description=entity_description, |
| 320 | source_id=entity_source_id, |
| 321 | ) |
| 322 | |
| 323 | |
| 324 | async def _handle_single_relationship_extraction( |
nothing calls this directly
no test coverage detected