Chain Modular

┌─────────────────────────────────────────────────────────────────────┐
│                          chain framework                            │
│                                                                     │
│  ┌─────────────────────────────────────────────────────────────┐   │
│  │  core/                                                       │   │
│  │  BaseChain ← SequentialChain, ParallelChain,                │   │
│  │              ConditionalChain, TransformChain,               │   │
│  │              CachingChain                                   │   │
│  │                                                              │   │
│  │  ChainConfig  ChainContext  ChainInput  ChainOutput          │   │
│  │  ExecutionTracer  TraceSpan                                  │   │
│  └─────────────────────────────────────────────────────────────┘   │
│                                                                     │
│  ┌──────────────────┐  ┌──────────────────┐  ┌─────────────────┐  │
│  │  chains/         │  │  registry/       │  │  utils/         │  │
│  │  SequentialChain │  │  ChainRegistry   │  │  cached         │  │
│  │  ParallelChain   │  │  register()      │  │  CachingChain   │  │
│  │  ConditionalChain│  │  get() / build() │  │  Declarative    │  │
│  │  TransformChain  │  │  @chain decorator│  │  ChainBuilder   │  │
│  └──────────────────┘  └──────────────────┘  └─────────────────┘  │
└─────────────────────────────────────────────────────────────────────┘

from fennec_community.chain import (
    TransformChain, SequentialChain, ParallelChain,
    ConditionalChain, ChainConfig, ChainContext,
    CachingChain, ChainRegistry, DeclarativeChainBuilder
)
import asyncio

# 1. Simple transform pipeline using >> operator
strip  = TransformChain(str.strip,  name="strip")
upper  = TransformChain(str.upper,  name="upper")
pipeline = strip >> upper

result = asyncio.run(pipeline.arun("  hello world  "))
print(result)  # "HELLO WORLD"

# 2. Parallel fan-out
parallel = ParallelChain({
    "stripped": strip,
    "uppercased": upper,
})
result = asyncio.run(parallel.arun("  hello  "))
print(result)  # {"stripped": "hello", "uppercased": "  HELLO  "}

# 3. Conditional routing
def router(data): return "upper" if data.islower() else "strip"
conditional = ConditionalChain(
    condition=router,
    branches={"upper": upper, "strip": strip}
)
result = asyncio.run(conditional.arun("hello"))
print(result)  # "HELLO"

from fennec_community.chain import ChainConfig

cfg = ChainConfig(
    retries=3,
    retry_delay=1.0,
    timeout=10.0,
    fallback_chain=my_backup_chain,
    verbose=True
)

from fennec_community.chain import ChainInput

inp = ChainInput(
    data={"query": "hello"},
    metadata={"user_id": "u123"},
)
inp.context["intent"] = "greet"

from fennec_community.chain import ChainOutput

out = ChainOutput(
    data={"result": "HELLO"},
    success=True,
    execution_time=0.003
)

from fennec_community.chain import ChainContext

ctx = ChainContext(session_id="sess-001", user_role="admin")
ctx["intent"] = "purchase"
ctx.set("cart_total", 99.9, namespace="shop")

print(ctx["intent"])                            # "purchase"
print(ctx.get("cart_total", namespace="shop"))  # 99.9
print(ctx.get("missing", default=0))            # 0

ChainContext(session_id: Optional[str] = None, **initial_values: Any)

get(key: str, default: Any = None, *, namespace: Optional[str] = None) -> Any

ctx.set("price", 49.99, namespace="checkout")
price = ctx.get("price", namespace="checkout")   # 49.99
missing = ctx.get("price")                       # None — not in default ns

set(key: str, value: Any, *, namespace: Optional[str] = None) -> None

ctx.set("token", "abc123", namespace="auth")
ctx.set("token", "xyz789", namespace="api")
# Both coexist without collision
print(ctx.get("token", namespace="auth"))  # "abc123"
print(ctx.get("token", namespace="api"))   # "xyz789"

update(mapping: Dict[str, Any]) -> None

ctx.update({"step": "payment", "amount": 120.0, "currency": "USD"})

as_dict() -> Dict[str, Any]

snapshot = ctx.as_dict()
print(json.dumps(snapshot))  # safe to serialise

to_dict() -> Dict[str, Any]

import json
trace_dict = tracer.root.to_dict()
print(json.dumps(trace_dict, indent=2))

pretty(indent: int = 0) -> str

print(tracer.root.pretty())
# ✅ [SequentialChain] 12.3ms
#     ✅ [strip] 0.1ms
#     ✅ [upper] 0.2ms

from fennec_community.chain import ExecutionTracer

tracer = ExecutionTracer()
with tracer.span("MyChain") as span:
    span.set_input("hello")
    result = do_something()
    span.set_output(result)

print(tracer.pretty())

@contextmanager
span(chain_name: str) -> Generator[TraceSpan, None, None]

tracer = ExecutionTracer()
with tracer.span("Pipeline") as root_span:
    root_span.set_input("raw input")
    with tracer.span("Step1") as child_span:
        child_span.set_output("step1 result")
    root_span.set_output("final result")

print(tracer.pretty())
# ✅ [Pipeline] 5.2ms
#     ✅ [Step1] 1.1ms

to_dict() -> Dict[str, Any]

import json
print(json.dumps(tracer.to_dict(), indent=2))

pretty() -> str

print(tracer.pretty())
# ✅ [SequentialChain] 14.7ms
#     ✅ [strip_whitespace] 0.2ms
#     ✅ [to_upper] 0.1ms
#     ✅ [ParallelChain] 8.3ms
#         ✅ [summary_chain] 7.1ms
#         ✅ [sentiment_chain] 6.8ms

from fennec_community.chain import BaseChain, ChainInput, ChainOutput, ChainConfig

class MyLLMChain(BaseChain):
    def __init__(self, llm, **kwargs):
        super().__init__(name="LLMChain", config=ChainConfig(retries=2, timeout=30.0), **kwargs)
        self.llm = llm

    async def _aexecute(self, inp: ChainInput) -> ChainOutput:
        answer = await self.llm.agenerate(inp.data)
        return ChainOutput(data=answer, success=True)

BaseChain(
    name: Optional[str] = None,
    config: Optional[ChainConfig] = None,
    tracer: Optional[ExecutionTracer] = None
)

run(input_data: Any, **kwargs: Any) -> Any

result = my_chain.run("hello world", source="api", user_id="u42")
print(result)  # processed output

async arun(input_data: Any, **kwargs: Any) -> Any

# In an async function
result = await my_chain.arun("hello world")

# As the entry point of a script
import asyncio
result = asyncio.run(my_chain.arun("hello world"))

__call__(input_data: Any, **kwargs: Any) -> Any

# These three are equivalent:
result = my_chain.run("data")
result = my_chain("data")
result = asyncio.run(my_chain.arun("data"))

__rshift__(other: BaseChain) -> SequentialChain

step1 = TransformChain(str.strip)
step2 = TransformChain(str.upper)
step3 = TransformChain(str.split)

pipeline = step1 >> step2 >> step3
# equivalent to SequentialChain([step1, step2, step3])

result = asyncio.run(pipeline.arun("  hello world  "))
# ["HELLO", "WORLD"]

__or__(other: BaseChain) -> SequentialChain

pipeline = step1 | step2 | step3

get_stats() -> Dict[str, Any]

stats = my_chain.get_stats()
print(stats)
# {
#   'chain_id': 'MyChain-a1b2c3d4',
#   'name': 'MyChain',
#   'total_executions': 100,
#   'success_count': 98,
#   'error_count': 2,
#   'success_rate': 0.98,
#   'avg_execution_time_ms': 42.7
# }

get_trace() -> Dict[str, Any]

import json

await my_pipeline.arun("test input")
trace = my_pipeline.get_trace()
print(json.dumps(trace, indent=2))
# {
#   "span_id": "a1b2c3d4",
#   "chain": "SequentialChain",
#   "duration_ms": 12.3,
#   "success": true,
#   "steps": [
#     {"span_id": "...", "chain": "strip", ...},
#     {"span_id": "...", "chain": "upper", ...}
#   ]
# }

from fennec_community.chain import SequentialChain, TransformChain
import asyncio

pipeline = SequentialChain([
    TransformChain(str.strip),
    TransformChain(str.upper),
    TransformChain(str.split),
], name="text_pipeline")

result = asyncio.run(pipeline.arun("  hello world  "))
print(result)  # ["HELLO", "WORLD"]

SequentialChain(
    chains: List[BaseChain],
    name: str = "SequentialChain",
    config: Optional[ChainConfig] = None,
    tracer: Optional[ExecutionTracer] = None
)

add(chain: BaseChain) -> "SequentialChain"

pipeline = SequentialChain([])
pipeline.add(step1).add(step2).add(step3)

# Equivalent to:
pipeline = SequentialChain([step1, step2, step3])

from fennec_community.chain import ParallelChain, TransformChain
import asyncio

parallel = ParallelChain({
    "original": TransformChain(lambda x: x),
    "upper":    TransformChain(str.upper),
    "words":    TransformChain(str.split),
})

result = asyncio.run(parallel.arun("hello world"))
print(result)
# {
#   "original": "hello world",
#   "upper": "HELLO WORLD",
#   "words": ["hello", "world"]
# }

ParallelChain(
    chains: Union[Dict[str, BaseChain], List[BaseChain]],
    name: str = "ParallelChain",
    fail_fast: bool = True,
    config: Optional[ChainConfig] = None,
    tracer: Optional[ExecutionTracer] = None
)

parallel = ParallelChain(
    {"fast": fast_chain, "slow": potentially_failing_chain},
    fail_fast=False
)
result = await parallel.arun("data")
# result.data = {"fast": "ok", "slow": None}
# result.metadata["branch_errors"] = {"slow": "timeout"}

from fennec_community.chain import ConditionalChain, TransformChain
import asyncio

def router(data: dict) -> str:
    return "high" if data.get("score", 0) > 0.7 else "low"

chain = ConditionalChain(
    condition=router,
    branches={
        "high": TransformChain(lambda d: f"HIGH PRIORITY: {d}"),
        "low":  TransformChain(lambda d: f"Normal: {d}"),
    },
    default=TransformChain(lambda d: f"Unknown: {d}")
)

result = asyncio.run(chain.arun({"score": 0.9, "text": "urgent"}))
print(result)  # "HIGH PRIORITY: {'score': 0.9, 'text': 'urgent'}"

ConditionalChain(
    condition: Callable[[Any], str],
    branches: Dict[str, BaseChain],
    default: Optional[BaseChain] = None,
    name: str = "ConditionalChain",
    config: Optional[ChainConfig] = None,
    tracer: Optional[ExecutionTracer] = None
)

from fennec_community.chain import TransformChain
import json, asyncio

parse_json  = TransformChain(json.loads,    name="json_parser")
to_upper    = TransformChain(str.upper,     name="to_upper")
word_count  = TransformChain(lambda s: len(s.split()), name="word_count")

pipeline = parse_json >> TransformChain(lambda d: d["text"]) >> to_upper
result = asyncio.run(pipeline.arun('{"text": "hello world"}'))
print(result)  # "HELLO WORLD"

TransformChain(
    transform_fn: Callable[[Any], Any],
    name: Optional[str] = None,
    config: Optional[ChainConfig] = None,
    tracer: Optional[ExecutionTracer] = None
)

from fennec_community.chain import ChainRegistry

# Look up and instantiate a registered chain
cls = ChainRegistry.get("transform")
chain = cls(str.upper)

# Or in one call
chain = ChainRegistry.build("transform", str.upper)

@classmethod
register(alias: str, chain_cls: Type[BaseChain]) -> None

class MyLLMChain(BaseChain):
    ...

ChainRegistry.register("llm", MyLLMChain)

@classmethod
chain(alias: str) -> Callable[[Type[BaseChain]], Type[BaseChain]]

@ChainRegistry.chain("sentiment")
class SentimentChain(BaseChain):
    async def _aexecute(self, inp):
        score = my_model.predict(inp.data)
        return ChainOutput(data=score, success=True)

# Now accessible:
chain = ChainRegistry.build("sentiment")

@classmethod
get(alias: str) -> Type[BaseChain]

TransformCls = ChainRegistry.get("transform")
chain = TransformCls(str.upper, name="my_upper")

@classmethod
build(alias: str, *args: Any, **kwargs: Any) -> BaseChain

transform_chain = ChainRegistry.build("transform", str.strip, name="strip_ws")
seq = ChainRegistry.build("sequential", [step1, step2], name="my_pipeline")

@classmethod
list_all() -> Dict[str, str]

print(ChainRegistry.list_all())
# {
#   'sequential':  'SequentialChain',
#   'parallel':    'ParallelChain',
#   'conditional': 'ConditionalChain',
#   'transform':   'TransformChain',
#   'sentiment':   'SentimentChain',   ← custom registration
# }

@classmethod
unregister(alias: str) -> None

@classmethod
clear() -> None

cached(fn: Callable) -> Callable

from fennec_community.chain import cached

@cached
def expensive_lookup(query: str) -> str:
    return slow_database.fetch(query)

result1 = expensive_lookup("hello")  # executes the function
result2 = expensive_lookup("hello")  # returns from cache ⚡

expensive_lookup.cache_clear()  # wipes all cached entries for this function

from fennec_community.chain import CachingChain

slow_chain = MyExpensiveLLMChain()
cached_chain = CachingChain(slow_chain)

result1 = await cached_chain.arun("What is AI?")  # executes slow_chain
result2 = await cached_chain.arun("What is AI?")  # returns from cache ⚡

print(cached_chain.cache_stats)
# {"hits": 1, "misses": 1, "size": 1}

CachingChain(
    chain: BaseChain,
    name: Optional[str] = None,
    config: Optional[ChainConfig] = None,
    tracer: Optional[ExecutionTracer] = None
)

cache_clear() -> None

cached_chain.cache_clear()
# All subsequent calls will miss the cache and execute the inner chain

@property
cache_stats -> Dict[str, int]

stats = cached_chain.cache_stats
print(f"Hit rate: {stats['hits'] / (stats['hits'] + stats['misses']):.1%}")

from fennec_community.chain import DeclarativeChainBuilder
import asyncio

spec = {
    "type": "sequential",
    "name": "text_pipeline",
    "steps": [
        {"type": "transform", "fn_name": "strip"},
        {"type": "transform", "fn_name": "upper"},
    ]
}

pipeline = DeclarativeChainBuilder.from_dict(
    spec,
    fn_registry={"strip": str.strip, "upper": str.upper}
)

result = asyncio.run(pipeline.arun("  hello world  "))
print(result)  # "HELLO WORLD"

@staticmethod
from_dict(
    spec: Dict[str, Any],
    fn_registry: Optional[Dict[str, Callable]] = None,
    chain_registry: Optional[Dict[str, Type[BaseChain]]] = None
) -> BaseChain

# Sequential pipeline
{
    "type": "sequential",
    "name": "my_pipeline",       # optional
    "steps": [                   # list of nested specs
        {"type": "transform", "fn_name": "strip"},
        {"type": "transform", "fn_name": "upper"},
    ]
}

# Parallel fan-out (named branches → dict result)
{
    "type": "parallel",
    "branches": {
        "summary":   {"type": "transform", "fn_name": "summarize"},
        "sentiment": {"type": "transform", "fn_name": "analyze"},
    }
}

# Parallel fan-out (list branches → list result)
{
    "type": "parallel",
    "branches": [
        {"type": "transform", "fn_name": "fn_a"},
        {"type": "transform", "fn_name": "fn_b"},
    ]
}

# Transform step — callable by value
{"type": "transform", "fn": my_callable_object}

# Transform step — callable by name from fn_registry
{"type": "transform", "fn_name": "my_fn", "name": "optional_label"}

# Any alias registered in ChainRegistry
{"type": "sentiment", "name": "sentiment_step"}

spec = {
    "type": "sequential",
    "steps": [
        {"type": "transform", "fn_name": "strip"},
        {
            "type": "parallel",
            "branches": {
                "upper": {"type": "transform", "fn_name": "upper"},
                "length": {"type": "transform", "fn_name": "length"},
            }
        }
    ]
}

pipeline = DeclarativeChainBuilder.from_dict(spec, fn_registry={
    "strip":  str.strip,
    "upper":  str.upper,
    "length": len,
})

result = asyncio.run(pipeline.arun("  hello  "))
print(result)  # {"upper": "HELLO", "length": 5}

@staticmethod
from_json(
    json_str: str,
    fn_registry: Optional[Dict[str, Callable]] = None,
    chain_registry: Optional[Dict[str, Type[BaseChain]]] = None
) -> BaseChain

json_spec = '''
{
  "type": "sequential",
  "steps": [
    {"type": "transform", "fn_name": "strip"},
    {"type": "transform", "fn_name": "upper"}
  ]
}
'''

pipeline = DeclarativeChainBuilder.from_json(
    json_spec,
    fn_registry={"strip": str.strip, "upper": str.upper}
)
result = asyncio.run(pipeline.arun("  hello  "))
print(result)  # "HELLO"

@staticmethod
from_yaml(
    yaml_str: str,
    fn_registry: Optional[Dict[str, Callable]] = None,
    chain_registry: Optional[Dict[str, Type[BaseChain]]] = None
) -> BaseChain

yaml_spec = """
type: sequential
name: text_pipeline
steps:
  - type: transform
    fn_name: strip
  - type: parallel
    branches:
      upper:
        type: transform
        fn_name: upper
      words:
        type: transform
        fn_name: split
"""

pipeline = DeclarativeChainBuilder.from_yaml(
    yaml_spec,
    fn_registry={
        "strip": str.strip,
        "upper": str.upper,
        "split": str.split,
    }
)

result = asyncio.run(pipeline.arun("  hello world  "))
print(result)  # {"upper": "HELLO WORLD", "words": ["hello", "world"]}

from chain import ChainConfig

# 3 retries, 1-second delay between attempts
cfg = ChainConfig(retries=3, retry_delay=1.0)
chain = MyChain(config=cfg)

cfg = ChainConfig(timeout=5.0)  # fail after 5 seconds
chain = MyChain(config=cfg)

from fennec_community.chain import ChainConfig, TransformChain

fallback = TransformChain(lambda _: "default answer", name="fallback")
cfg = ChainConfig(retries=2, fallback_chain=fallback)
primary = MyUnreliableChain(config=cfg)

# If primary fails 3 times, fallback chain is called with the original input
result = await primary.arun("query")

parallel = ParallelChain(
    {"branch_a": chain_a, "branch_b": chain_b},
    fail_fast=False  # collect all results, don't abort on first failure
)
output = await parallel.arun("data")

if output.metadata.get("branch_errors"):
    print("Some branches failed:", output.metadata["branch_errors"])

import asyncio
from fennec_community.chain import BaseChain, ChainInput, ChainOutput, ChainConfig, ExecutionTracer

class LLMChain(BaseChain):
    def __init__(self, llm, **kwargs):
        super().__init__(
            name="LLMChain",
            config=ChainConfig(retries=2, timeout=30.0, verbose=True),
            **kwargs
        )
        self.llm = llm

    async def _aexecute(self, inp: ChainInput) -> ChainOutput:
        answer = await self.llm.agenerate(inp.data)
        return ChainOutput(data=answer, success=True)

llm_chain = LLMChain(my_llm)
result = asyncio.run(llm_chain.arun("What is machine learning?"))
print(result)

# Observability
print(llm_chain.get_stats())
import json
print(json.dumps(llm_chain.get_trace(), indent=2))

import asyncio
from fennec_community.chain import TransformChain, SequentialChain, ChainInput, ChainContext

# Build context
ctx = ChainContext(session_id="session-001")
ctx["user_tier"] = "premium"

# Build pipeline
pipeline = (
    TransformChain(str.strip,  name="strip")
    >> TransformChain(str.lower, name="lowercase")
    >> TransformChain(str.split, name="tokenize")
)

# Pass context via ChainInput
inp = ChainInput(data="  Hello World  ", context=ctx)
result = asyncio.run(pipeline.arun(inp))
print(result)  # ["hello", "world"]

# Shared context is available to all steps
print(ctx["user_tier"])  # "premium" — still accessible

import asyncio
from fennec_community.chain import TransformChain, ParallelChain, SequentialChain

# Stage 1: parallel enrichment
enrich = ParallelChain({
    "upper":  TransformChain(str.upper),
    "words":  TransformChain(str.split),
    "length": TransformChain(len),
})

# Stage 2: merge results
def merge(d): return f"TEXT: {d['upper']} | WORDS: {d['words']} | LEN: {d['length']}"
merge_chain = TransformChain(merge, name="merge")

pipeline = enrich >> merge_chain
result = asyncio.run(pipeline.arun("hello world"))
print(result)
# "TEXT: HELLO WORLD | WORDS: ['hello', 'world'] | LEN: 11"

import asyncio
from fennec_community.chain import ConditionalChain, TransformChain, SequentialChain

def classify(data: str) -> str:
    return "question" if data.strip().endswith("?") else "statement"

question_chain  = TransformChain(lambda t: f"[Q] {t}", name="question_handler")
statement_chain = TransformChain(lambda t: f"[S] {t}", name="statement_handler")

router = ConditionalChain(
    condition=classify,
    branches={"question": question_chain, "statement": statement_chain},
    default=TransformChain(lambda t: f"[?] {t}", name="unknown_handler")
)

pipeline = TransformChain(str.strip) >> router

print(asyncio.run(pipeline.arun("What is AI?  ")))   # [Q] What is AI?
print(asyncio.run(pipeline.arun("AI is powerful."))) # [S] AI is powerful.

import asyncio
from fennec_community.chain import CachingChain

expensive_chain = MyLLMChain()
cached = CachingChain(expensive_chain)

q = "Explain quantum computing"
r1 = asyncio.run(cached.arun(q))  # executes LLM
r2 = asyncio.run(cached.arun(q))  # cache hit ⚡
r3 = asyncio.run(cached.arun(q))  # cache hit ⚡

print(cached.cache_stats)  # {"hits": 2, "misses": 1, "size": 1}

cached.cache_clear()  # force refresh
r4 = asyncio.run(cached.arun(q))  # executes LLM again

import asyncio
from fennec_community.chain import ChainRegistry, DeclarativeChainBuilder, BaseChain, ChainInput, ChainOutput

# Register a custom chain
@ChainRegistry.chain("word_count")
class WordCountChain(BaseChain):
    async def _aexecute(self, inp: ChainInput) -> ChainOutput:
        return ChainOutput(data=len(inp.data.split()), success=True)

# Build from YAML config (loaded from file / env / API)
yaml_config = """
type: sequential
steps:
  - type: transform
    fn_name: strip
  - type: word_count
"""

pipeline = DeclarativeChainBuilder.from_yaml(
    yaml_config,
    fn_registry={"strip": str.strip}
)

result = asyncio.run(pipeline.arun("  hello beautiful world  "))
print(result)   # 3
print(ChainRegistry.list_all())

from fastapi import FastAPI
from fennec_community.chain import SequentialChain, TransformChain, CachingChain, ChainConfig
import asyncio

app = FastAPI()

pipeline = CachingChain(
    SequentialChain([
        TransformChain(str.strip, name="strip"),
        TransformChain(str.upper, name="upper"),
    ], config=ChainConfig(retries=2, timeout=5.0)),
)

@app.post("/process")
async def process(text: str):
    result = await pipeline.arun(text)
    return {
        "result": result,
        "cache_stats": pipeline.cache_stats,
        "chain_stats": pipeline.get_stats(),
    }

@app.get("/trace")
async def get_trace():
    return pipeline.get_trace()