# ParallelEnrichment

> **Class** in `langchain_parallel`

📖 [View in docs](https://reference.langchain.com/python/langchain-parallel/tasks/ParallelEnrichment)

High-level Runnable for the structured-batch enrichment pattern.

Wraps :class:`ParallelTaskGroup` with a ``default_task_spec`` so callers
can pass a list of records (pydantic instances or dicts) and get back a
list of enriched output dicts. Mirrors the canonical enrichment pattern
documented at
https://docs.parallel.ai/task-api/examples/task-enrichment.

Defaults to the ``core`` processor (the docs' recommendation for
enrichment workflows).

## Signature

```python
ParallelEnrichment(
    self,
    *,
    output_schema: Union[type[BaseModel], dict[str, Any], str],
    input_schema: Union[type[BaseModel], dict[str, Any], str, None] = None,
    processor: ProcessorLiteral = 'core-fast',
    api_key: Optional[Union[str, SecretStr]] = None,
    base_url: str = 'https://api.parallel.ai',
    metadata: Optional[dict[str, Union[str, float, bool]]] = None,
)
```

## Description

**Example:**

```python
from pydantic import BaseModel, Field
from langchain_parallel import ParallelEnrichment

class CompanyInput(BaseModel):
    company: str = Field(description="Company name to enrich")

class CompanyOutput(BaseModel):
    headquarters: str
    founding_year: int = Field(description="Year the company was founded")

enricher = ParallelEnrichment(
    input_schema=CompanyInput,
    output_schema=CompanyOutput,
    # Defaults to processor="core-fast"; pass "core" or "pro" for
    # higher accuracy when latency is less of a concern.
)

results = enricher.invoke([
    {"company": "Anthropic"},
    {"company": "OpenAI"},
    {"company": "Google DeepMind"},
])

for inp, out in zip(["Anthropic", "OpenAI", "Google DeepMind"], results):
    content = out["output"]["content"]
    print(inp, "->", content)
```

## Extends

- `Runnable[list[EnrichmentInput], list[dict[str, Any]]]`

## Constructors

```python
__init__(
    self,
    *,
    output_schema: Union[type[BaseModel], dict[str, Any], str],
    input_schema: Union[type[BaseModel], dict[str, Any], str, None] = None,
    processor: ProcessorLiteral = 'core-fast',
    api_key: Optional[Union[str, SecretStr]] = None,
    base_url: str = 'https://api.parallel.ai',
    metadata: Optional[dict[str, Union[str, float, bool]]] = None,
) -> None
```

| Name | Type |
|------|------|
| `output_schema` | `Union[type[BaseModel], dict[str, Any], str]` |
| `input_schema` | `Union[type[BaseModel], dict[str, Any], str, None]` |
| `processor` | `ProcessorLiteral` |
| `api_key` | `Optional[Union[str, SecretStr]]` |
| `base_url` | `str` |
| `metadata` | `Optional[dict[str, Union[str, float, bool]]]` |


## Properties

- `processor`

## Methods

- [`invoke()`](https://reference.langchain.com/python/langchain-parallel/tasks/ParallelEnrichment/invoke)
- [`ainvoke()`](https://reference.langchain.com/python/langchain-parallel/tasks/ParallelEnrichment/ainvoke)

---

[View source on GitHub](https://github.com/parallel-web/langchain-parallel/blob/c1f8c1d657b86eaf948c363f84fed6ea6bd65754/langchain_parallel/tasks.py#L867)