# Implementing a Custom Workflow in Heurist Agent Framework: A Complete Guide

> Learn how to implement a custom workflow in Heurist Agent Framework. This guide covers creating Python classes, handling LLM calls, and integrating workflows with agents. Get started today.

- Repository: [Heurist/heurist-agent-framework](https://github.com/heurist-network/heurist-agent-framework)
- Tags: how-to-guide
- Published: 2026-03-03

---

**Implementing a custom workflow in the Heurist Agent Framework requires creating a Python class with a standard constructor accepting `llm_provider` and `tool_manager`, implementing an async `process` method that handles option merging, prompt preparation, LLM calls, and result synthesis, then exporting the class in [`core/workflows/__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py) for integration with agents.**

The Heurist Agent Framework provides a modular architecture for building complex, multi-step LLM orchestrations through reusable workflow classes. Whether you need to implement chain-of-thought reasoning, deep research capabilities, or domain-specific processing pipelines, understanding the process for implementing a custom workflow allows you to extend the framework's capabilities while maintaining compatibility with the CoreAgent and tool management systems.

## Understanding the Workflow Architecture

The framework ships with two reference implementations that demonstrate the standard pattern for implementing a custom workflow:

| Workflow | Core File | Primary Responsibilities |
|----------|-----------|--------------------------|
| **Chain-of-Thought Reasoning** | [`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py) | Generates a planning JSON, executes each step (optionally via tools), and builds a final answer. |
| **Research Workflow** | [`core/workflows/deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/deep_research.py) | Generates search queries, performs parallel web searches (single- or multi-provider), extracts learnings, recursively explores follow-up questions, and produces a structured research report. |

Both implementations follow a consistent six-phase execution pattern:

1. **Receive the user message** (and optional `personality_provider`, `chat_id`, etc.).
2. **Merge default workflow options with caller-provided overrides**.
3. **Run a dedicated LLM step** (planning, query generation, or prompt synthesis).
4. **Iterate over sub-steps** – calling tools, performing searches, or invoking other LLM calls.
5. **Collect intermediate results** (`steps_responses`, `learnings`, `analyses`).
6. **Produce a final response** (optionally with a report or formatted output).

## Step-by-Step Guide to Implementing a Custom Workflow

### 1. Create a New Workflow Class

Create a Python module under `core/workflows/` (e.g., [`my_custom_workflow.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/my_custom_workflow.py)). Start with the same constructor signature used by the built-ins:

```python
class MyCustomWorkflow:
    """Custom workflow – replace with a descriptive docstring."""

    def __init__(self, llm_provider, tool_manager, **extra):
        # Store the LLM and tool manager – they are required for every step

        self.llm_provider = llm_provider
        self.tool_manager = tool_manager
        # Accept any extra dependencies (search clients, vector stores, …) via **extra

        for k, v in extra.items():
            setattr(self, k, v)

```

This pattern matches the initialization logic found in `ChainOfThoughtReasoning` ([`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py), lines 12-15) and `ResearchWorkflow` ([`core/workflows/deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/deep_research.py), lines 29-38).

### 2. Define the Async `process` Method

All workflows expose a single async method named `process` with the signature:

```python
async def process(
    self,
    message: str,
    personality_provider=None,
    chat_id: str = None,
    workflow_options: Dict = None,
    **kwargs,
) -> Tuple[Optional[str], Optional[str], Optional[Dict]]:
    ...

```

Copy this signature verbatim; the framework expects it when a workflow is passed to an agent (e.g., `Agent(..., workflow=MyCustomWorkflow(...))`).

Inside the method, implement these phases:

**Option Handling**

Initialize a defaults dictionary, then merge `workflow_options`:

```python
options = {"temperature": 0.7, "use_tools": False}
if workflow_options:
    options.update(workflow_options)

```

**Prompt Preparation**

Build a system/user prompt that tells the LLM what to do (planning, query generation, etc.):

```python
system_prompt = "You are a planning assistant. Generate a JSON plan..."
user_prompt = f"User request: {message}"

```

Reference the planning prompt construction in `ChainOfThoughtReasoning` ([`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py), lines 48-64).

**LLM Call**

Use `await self.llm_provider.call(...)` with appropriate parameters:

```python
text_response, _, _ = await self.llm_provider.call(
    system_prompt=system_prompt,
    user_prompt=user_prompt,
    temperature=options["temperature"],
    skip_tools=not options["use_tools"]
)

```

**Result Parsing**

Parse JSON or plain-text results; include fallback handling for malformed output:

```python
try:
    plan = json.loads(text_response)
except json.JSONDecodeError:
    plan = self._json_fallback(text_response)

```

Reference the `_json_fallback` method in `ChainOfThoughtReasoning` ([`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py), lines 105-138).

**Iterative Sub-steps**

Loop over the parsed plan, optionally calling tools:

```python
for step in plan["steps"]:
    if options["use_tools"]:
        tools_config = self.tool_manager.get_tools_config()
    # Execute step logic...

```

See the execution loop starting at line 173 in `ChainOfThoughtReasoning`.

**Final Synthesis**

Build a concluding prompt and call the LLM again for the final answer:

```python
final_prompt = f"Based on these results: {intermediate_results}, provide a final answer."
final_response, _, _ = await self.llm_provider.call(...)

```

**Error Handling**

Wrap the whole flow in a `try/except` and fall back to a simple direct LLM call if anything crashes:

```python
try:
    # ... full workflow logic ...

except Exception as e:
    # Fallback to direct LLM call

    return await self.llm_provider.call(...)

```

Reference the error handling block at lines 200-203 in `ChainOfThoughtReasoning`.

### 3. Add Helper Methods for Reusable Logic

Complex workflows often factor out reusable pieces. Add any helpers you need, following the same async signature and returning plain Python data structures:

- `_generate_questions` – creates clarifying questions (used by `ResearchWorkflow`)
- `_generate_search_queries` – builds intelligent SERP queries (used by `ResearchWorkflow`)
- `_process_search_result` – extracts learnings from raw search data
- `_json_fallback` – attempts to repair malformed JSON (used by `ChainOfThoughtReasoning`)
- `_fallback` – simple direct LLM call when the full pipeline fails

### 4. Export the Workflow Class

Expose your class via the package's public API in [`core/workflows/__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py):

```python
from .my_custom_workflow import MyCustomWorkflow

__all__ = [
    "AugmentedLLMCall",
    "ChainOfThoughtReasoning",
    "ResearchWorkflow",
    "MyCustomWorkflow",          # ← add this line

]

```

Reference the current export list at [`core/workflows/__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py) (lines 9-10).

### 5. Integrate with an Agent

Any agent can now be instantiated with the new workflow. For example, in a test script or an application entry point:

```python
from core.workflows import MyCustomWorkflow
from agents.core_agent import CoreAgent   # example agent

from llm import MyLLMProvider               # your concrete LLM provider

from tools.tool_box import DefaultToolBox   # tool manager

llm = MyLLMProvider(...)
tools = DefaultToolBox(...)
workflow = MyCustomWorkflow(llm, tools)

agent = CoreAgent(
    llm_provider=llm,
    tool_manager=tools,
    workflow=workflow,
    # other agent options …

)

response, image, extra = await agent.handle_message("Explain the impact of quantum computing on finance")
print(response)

```

Reference the agent integration pattern in [`agents/core_agent.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/agents/core_agent.py) (line 539 shows `research_workflow = ResearchWorkflow(...)`).

### 6. Test with a Standalone Script

Place a runnable example under `core/examples/` or `mesh/test_scripts/` so future developers can see the workflow in action:

```python

# core/examples/my_custom_workflow_demo.py

import asyncio
from core.workflows import MyCustomWorkflow
from llm import MyLLMProvider
from tools.tool_box import DefaultToolBox

async def main():
    llm = MyLLMProvider()
    tools = DefaultToolBox()
    workflow = MyCustomWorkflow(llm, tools)
    response, _, _ = await workflow.process(
        "Summarize the latest developments in AI alignment",
        workflow_options={"temperature": 0.5, "use_tools": True},
    )
    print(response)

if __name__ == "__main__":
    asyncio.run(main())

```

Running it with `uv run python core/examples/my_custom_workflow_demo.py` will demonstrate the full end-to-end flow.

## Code Examples for Common Workflow Patterns

### Chain-of-Thought Reasoning Implementation

```python
from core.workflows import ChainOfThoughtReasoning
from llm import OpenAIProvider
from tools.tool_box import DefaultToolBox

llm = OpenAIProvider(model="gpt-4o-mini")
tools = DefaultToolBox()
cot = ChainOfThoughtReasoning(llm, tools, augmented_llm=None)   # `augmented_llm` can be a wrapper that adds tool support

response, image_url, _ = await cot.process(
    "What are the steps to launch a DeFi token on Ethereum?",
    workflow_options={"temperature": 0.6, "use_tools": True},
)
print(response)

```

Key lines referenced: construction ([`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py), lines 12-15), call to `process` (lines 17-19), option overrides (lines 34-36).

### Research Workflow with Multi-Provider Search

```python
from core.workflows import ResearchWorkflow
from llm import OpenAIProvider
from tools.tool_box import DefaultToolBox
from core.clients.search_client import ExaClient, DuckDuckGoClient

llm = OpenAIProvider(model="gpt-4o")
tools = DefaultToolBox()
search_clients = {
    "exa": ExaClient(api_key="YOUR_EXA_API_KEY"),
    "duckduckgo": DuckDuckGoClient(),
}
research = ResearchWorkflow(llm, tools, search_clients=search_clients)

report, _, result = await research.process(
    "Analyze the recent regulatory landscape for stablecoins in the EU",
    workflow_options={
        "breadth": 2,
        "depth": 3,
        "multi_provider": True,
        "search_providers": ["exa", "duckduckgo"],
    },
)
print(report)               # markdown report

print(result["visited_urls"])  # list of source URLs

```

Key sections: initialization ([`core/workflows/deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/deep_research.py), lines 29-38), option handling (lines 86-99), multi-provider logic (lines 107-119).

### Custom Workflow Integration

```python
from agents.core_agent import CoreAgent
from core.workflows import MyCustomWorkflow
from llm import MyLLMProvider
from tools.tool_box import DefaultToolBox

llm = MyLLMProvider()
tools = DefaultToolBox()
my_wf = MyCustomWorkflow(llm, tools, extra_service=my_service)

agent = CoreAgent(
    llm_provider=llm,
    tool_manager=tools,
    workflow=my_wf,
    name="MySpecialAgent",
)

response, _, _ = await agent.handle_message("Give me a step-by-step plan for migrating my database to the cloud")
print(response)

```

## Key Source Files to Reference

| File | Purpose | Direct link |
|------|---------|-------------|
| [`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py) | Full implementation of the CoT reasoning flow (planning → execution → final synthesis). | [[`chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/chain_of_thought.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py) |
| [`core/workflows/deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/deep_research.py) | Comprehensive research pipeline (query generation, parallel searches, recursion, report creation). | [[`deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/deep_research.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/deep_research.py) |
| [`core/workflows/__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py) | Export list that makes workflow classes importable as `core.workflows`. | [[`__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/__init__.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py) |
| [`core/examples/deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/examples/deep_research.py) | Example script showing how to instantiate `ResearchWorkflow`. | [[`deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/deep_research.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/examples/deep_research.py) |
| [`core/examples/basic_calls.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/examples/basic_calls.py) | Demonstrates usage of `ChainOfThoughtReasoning` and the generic `AugmentedLLMCall`. | [[`basic_calls.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/basic_calls.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/examples/basic_calls.py) |
| [`agents/core_agent.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/agents/core_agent.py) | Core agent class that accepts a workflow object and routes messages through it. | [[`core_agent.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core_agent.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/agents/core_agent.py) |
| [`core/clients/search_client.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/clients/search_client.py) | Abstract search client interface used by `ResearchWorkflow`. | [[`search_client.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/search_client.py)](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/clients/search_client.py) |

## Summary

- **Create a class** with the same constructor pattern (`llm_provider`, `tool_manager`, `**extra`) as the built-ins.
- **Implement an async `process` method** that merges default options, builds prompts, calls the LLM (and tools if needed), iterates over sub-steps, and synthesizes a final response.
- **Add helper methods** for reusable logic such as JSON parsing fallbacks, question generation, or search query construction.
- **Export the class** in [`core/workflows/__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py) to make it available for import.
- **Instantiate the workflow** when constructing an agent or in standalone scripts, passing any required extra dependencies like search clients via the `**extra` parameter.

## Frequently Asked Questions

### What is the minimum required structure for a custom workflow class?

A custom workflow class must implement an `__init__` method that accepts `llm_provider` and `tool_manager` as positional arguments, plus `**extra` for additional dependencies. It must also expose an async `process` method with the signature `(message, personality_provider=None, chat_id=None, workflow_options=None, **kwargs)`. The constructor should store these dependencies as instance attributes so the `process` method can access them during execution.

### How do I handle errors when implementing a custom workflow?

Wrap the entire workflow logic in a `try/except` block within the `process` method. If an exception occurs, fall back to a simple direct LLM call using `self.llm_provider.call()` to ensure the user still receives a response. For JSON parsing errors specifically, implement a `_json_fallback` method that attempts to repair malformed JSON using regex or partial parsing, as demonstrated in [`core/workflows/chain_of_thought.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/chain_of_thought.py) (lines 105-138).

### Can I use multiple search providers in my custom workflow?

Yes, pass a dictionary of search clients via the `**extra` parameter in the constructor, then store them as instance attributes. In `process`, check `workflow_options` for `multi_provider` and `search_providers` keys to determine which clients to invoke. The `ResearchWorkflow` implementation demonstrates this pattern in [`core/workflows/deep_research.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/deep_research.py) (lines 107-119), where it iterates over specified providers and aggregates results from Exa, DuckDuckGo, or other configured clients.

### How do I register my custom workflow for use with CoreAgent?

After creating your workflow class, add it to the export list in [`core/workflows/__init__.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/core/workflows/__init__.py). Import the class at the top of the file (e.g., `from .my_custom_workflow import MyCustomWorkflow`), then append the class name to the `__all__` list. This makes the workflow available for import as `from core.workflows import MyCustomWorkflow`, which you can then pass to `CoreAgent` during instantiation via the `workflow` parameter, as shown in [`agents/core_agent.py`](https://github.com/heurist-network/heurist-agent-framework/blob/main/agents/core_agent.py) (line 539).