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

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 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	Generates a planning JSON, executes each step (optionally via tools), and builds a final answer.
Research Workflow	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). Start with the same constructor signature used by the built-ins:

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, lines 12-15) and ResearchWorkflow (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:

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:

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.):

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, lines 48-64).

LLM Call

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

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:

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, lines 105-138).

Iterative Sub-steps

Loop over the parsed plan, optionally calling tools:

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:

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:

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:

from .my_custom_workflow import MyCustomWorkflow

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

]

Reference the current export list at 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:

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 (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:

# 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

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, lines 12-15), call to process (lines 17-19), option overrides (lines 34-36).

Research Workflow with Multi-Provider Search

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, lines 29-38), option handling (lines 86-99), multi-provider logic (lines 107-119).

Custom Workflow Integration

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	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/core/workflows/chain_of_thought.py)
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/core/workflows/deep_research.py)
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/core/workflows/__init__.py)
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/core/examples/deep_research.py)
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/core/examples/basic_calls.py)
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/agents/core_agent.py)
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/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 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 (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 (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. 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 (line 539).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}