CSVOutputParser (csv_parser.py)

class output_parser.csv_parser.CSVOutputParser(BaseOutputParser)

Overview

Extracts a delimited table from raw LLM text and returns a list of dicts (one per data row), with optional required-column validation and per-column type coercion. Depends only on base_parser.py and the standard library (csv, io, re) — no external dependencies.

Constructor

CSVOutputParser(
    required_columns: Optional[List[str]] = None,
    column_types: Optional[Dict[str, type]] = None,
    delimiter: Optional[str] = None,
)

_DELIMITERS = [",", ";", "\t", "|"] — the fixed candidate set considered during auto-detection.

Public Methods

parse

def parse(self, text: str) -> List[Dict[str, Any]]

Returns: a list of dicts, one per data row, with lowercased/stripped column-name keys.

Raises:

• ParsingError("Cannot parse empty text", ...) for empty input.
• ParsingError(f"CSV parsing error: {e}", ...) if the standard-library csv module itself raises csv.Error.
• ParsingError("CSV contains no data rows.", ...) if the extracted text has a header but zero data rows.
• ParsingError(f"Missing required columns: {sorted(missing)}", ...) if required_columns were declared and any are absent from the header.
• ParsingError(f"Cannot coerce column '{col}' value {raw!r} to {type.__name__}: {e}", cause=e) — raised by _coerce_row if column_types coercion fails on any cell (bubbles up through parse() uncaught, since _coerce_row itself raises ParsingError directly rather than a bare exception).

Internal Workflow:

1. _extract_csv(text) — regex-strips a ```csv or bare ``` fence if present, else uses the whole stripped text.
2. Delimiter resolution: self.delimiter if explicitly set, else _detect_delimiter(csv_text).
3. csv.DictReader(io.StringIO(csv_text), delimiter=delimiter) — parses using the standard library's CSV dialect handling (quote handling, escaped delimiters, etc., all inherited from csv.DictReader's default dialect).
4. Normalizes every row: strips whitespace from and lowercases every header key, strips values ({k.strip().lower(): v.strip() for k, v in row.items() if k} — note: if k skips any column with a falsy/empty header name, e.g. from a trailing delimiter producing an unnamed column).
5. If required_columns set, checks set(required_columns) - set(rows[0].keys()) for missing columns — this check uses only the first row's keys as the reference for "actual columns", which is safe for well-formed CSV (DictReader guarantees consistent keys per row based on the header) but means a malformed row with extra/fewer fields elsewhere in the data won't be separately validated for column-set consistency here.
6. If column_types set, applies _coerce_row to every row.

Algorithm — delimiter detection (_detect_delimiter): counts occurrences of each of , ; \t | in the first line only, returns whichever has the highest count (defaulting to , if all counts are zero). This is simpler than format_detector.py's _score_csv (which checks row-to-row consistency across up to 4 rows) — CSVOutputParser's own detection only looks at the header line.

Complexity: O(n) in text length for extraction and delimiter counting; O(rows × columns) for row normalization and type coercion — standard for CSV parsing.

Thread Safety: Safe to share across threads for concurrent parse() calls — no mutable instance state beyond read-only config set at construction.

Side Effects: None beyond raising exceptions.

get_format_instructions

Returns instructions for a CSV table with a header row. If required_columns were declared, includes them explicitly as the required header along with a placeholder example row (value_col1,value_col2,...). Otherwise uses a generic column1,column2,column3 example. Always appends a note that a ```csv fence is optional but recommended.

Private Helpers

• _extract_csv(text) -> str
• _detect_delimiter(text) -> str
• _coerce_row(row: Dict[str, str]) -> Dict[str, Any] — for each cell, if its column has a declared type in column_types, applies self.column_types[col](raw); on ValueError/TypeError, raises ParsingError. Columns without a declared type pass through unchanged as strings.

Examples

# Basic
parser = CSVOutputParser()
text = "name,age,city\nAlice,30,Paris\nBob,25,London"
rows = parser.parse(text)
# [{'name': 'Alice', 'age': '30', 'city': 'Paris'}, {'name': 'Bob', 'age': '25', 'city': 'London'}]

# Intermediate — type coercion
parser = CSVOutputParser(column_types={"age": int})
rows = parser.parse(text)
# [{'name': 'Alice', 'age': 30, 'city': 'Paris'}, ...]   # age is now an int

# Advanced — required columns + fenced input
parser = CSVOutputParser(required_columns=["name", "age", "city"])
text = '''
```csv
name,age,city
Alice,30,Paris

''' rows = parser.parse(text)

parser = CSVOutputParser(required_columns=["name", "email"]) try: parser.parse("name,age\nAlice,30") except ParsingError as e: print(str(e)) # mentions missing 'email'

text = "name;age;city\nAlice;30;Paris" rows = CSVOutputParser().parse(text) # delimiter auto-detected as ';'

## Best Practices

- Set `column_types` for any numeric/boolean columns you plan to use programmatically — without it, every value is a plain `str`, including things that look like numbers.
- If you expect a specific delimiter (e.g. your prompt explicitly asks for tab-separated output), pass `delimiter="\t"` explicitly rather than relying on auto-detection, especially for single-column-count-ambiguous inputs.

## Common Mistakes

- Forgetting that `required_columns` and `column_types` keys are matched **case-insensitively via lowercasing** internally, but the returned row dicts also always use lowercased keys — if you access `rows[0]["Name"]` (capitalized) expecting it to work because you declared `required_columns=["Name"]`, it will `KeyError`; use `rows[0]["name"]`.
- Assuming `_detect_delimiter` looks at the whole file — it only inspects the **first line**. A CSV whose header happens to contain more commas in a quoted field than the intended delimiter's count elsewhere could, in rare cases, be misdetected. Prefer explicit `delimiter=` for atypical inputs.

## Related APIs

- [`BaseOutputParser`, `ParsingError`](base_parser.md)
- [`OutputParser`](output_parser.md) — the orchestrator's internal `_parse_csv` is a simpler variant without required-column/type-coercion support.
- [`FormatDetector`](format_detector.md) — `_score_csv` implements a more robust (multi-row) delimiter-consistency check than this class's own `_detect_delimiter`.

## Version History

Not determinable from source.