Workflow agent

this guide covers building multi agent workflows using bridgebaseworkflowagent and langgraph when to use bridgebaseworkflowagent use bridgebaseworkflowagent when your agent has multiple processing steps with conditional routing needs multiple specialized sub agents has complex state management requirements needs retry loops or fan out patterns base class overview from bridge agent sdk import bridgebaseworkflowagent, workflowconfig from langgraph graph import stategraph, end class myworkflow(bridgebaseworkflowagent) """your custom workflow agent """ config = workflowconfig( name="my workflow", 	 version="1 0 0", description="my workflow agent", ) def define state(self) > type """return the typeddict class defining workflow state """ return mystate def setup nodes(self, graph stategraph) """add nodes to the workflow graph """ pass def setup edges(self, graph stategraph) """define edges and routing between nodes """ pass lifecycle methods \ method \ \ purpose \ \ when called \ define state() define state schema (typeddict) on graph construction setup nodes() add processing nodes on graph construction setup edges() define flow routing on graph construction prepare initial state() build initial state from agentinput before graph execution extract output() extract output from final state after graph execution execute() orchestrates the full lifecycle on each invocation there is no build() or run() method the execute() method handles graph compilation and invocation automatically call await workflow\ execute(agent input) add workflow agent s tep 1 define workflow state create state in src/agents/state py """workflow state definitions """ from typing import typeddict, optional, list, dict, any, annotated from operator import add class workflowstatus """canonical workflow status values """ pending = "pending" in progress = "in progress" success = "success" error = "error" class workflowstate(typeddict, total=false) """state shared across all workflow nodes all nodes can read and write to this state langgraph manages state persistence and transitions """ \# input raw input dict platform context dict \# processing status str current node str \# results output optional\[str] error optional\[str] error message optional\[str] \# tracking (use annotated for append only lists) messages annotated\[list\[dict\[str, any]], add] step 2 create the workflow agent create agent in src/agents/orchestrator py """orchestrator workflow agent using langgraph """ import logging import os from typing import dict, any, literal from bridge agent sdk import ( bridgebaseworkflowagent, workflowconfig, agentinput, mcpclient, ) from langgraph graph import stategraph, end from src agents state import workflowstate, workflowstatus logger = logging getlogger( name ) class orchestratorworkflow(bridgebaseworkflowagent) """multi agent workflow that routes queries to specialized handlers flow 1\ classifier → determines query type (sql, rag, action) 2\ route to appropriate handler 3\ responder → formats final response """ config = workflowconfig( name="orchestrator", version="1 0 0", description="multi agent query orchestrator", ) classifier prompt = """classify the user query into one of these categories \ sql data queries, metrics, reports, statistics \ rag knowledge questions, how to, documentation respond with json {"type" "sql|rag", "confidence" 0 0 1 0}""" \# ───────────────────────────────────────────────────────────── \# required overrides \# ───────────────────────────────────────────────────────────── def define state(self) > type """define the state schema for this workflow """ return workflowstate def setup nodes(self, graph stategraph) """add processing nodes to the workflow graph """ graph add node("classifier", self classifier node) graph add node("sql agent", self sql agent node) graph add node("rag agent", self rag agent node) graph add node("responder", self responder node) def setup edges(self, graph stategraph) """define edges and routing between nodes """ graph set entry point("classifier") graph add conditional edges( "classifier", self route query, { "sql" "sql agent", "rag" "rag agent", 	 } ) graph add edge("sql agent", "responder") graph add edge("rag agent", "responder") graph add edge("responder", end) \# ───────────────────────────────────────────────────────────── \# node implementations \# ───────────────────────────────────────────────────────────── def classifier node(self, state workflowstate) > dict\[str, any] """classify the user query to determine routing """ logger info("classifying query ") \# self llm is auto created by the sdk's execute() method \# use self create llm(state) only if you need a fresh client \# with different settings (e g , different temperature) response = self llm invoke(\[ {"role" "system", "content" self classifier prompt}, {"role" "user", "content" str(state get("raw input", {}) get("query", ""))}, ]) classification = self parse classification(response content) return { "status" workflowstatus in progress, "current node" "classifier", "messages" \[{"role" "classifier", "content" response content}], } async def sql agent node(self, state workflowstate) > dict\[str, any] """process sql/data queries via mcp """ logger info("processing sql query ") platform context = state get("platform context", {}) mcp = mcpclient( dev mode=(os getenv("kaif mode") != "production"), agent payload=platform context if os getenv("kaif mode") != "local dev" else none, ) result = await mcp call tool parsed( "bridge execute query", {"query" 'select from "itsm" incident limit 10'}, ) success, parsed = result return { "output" str(parsed get("data", \[]) if success else \[]), "status" workflowstatus in progress, "current node" "sql agent", } def rag agent node(self, state workflowstate) > dict\[str, any] """process knowledge/rag queries """ logger info("processing rag query ") query = str(state get("raw input", {}) get("query", "")) response = self llm invoke(\[ {"role" "system", "content" "answer the question based on your knowledge "}, {"role" "user", "content" query}, ]) return { "output" response content, "status" workflowstatus in progress, "current node" "rag agent", } def responder node(self, state workflowstate) > dict\[str, any] """format the final response """ logger info("generating final response ") return { "status" workflowstatus success, "current node" "responder", } \# ───────────────────────────────────────────────────────────── \# routing function \# ───────────────────────────────────────────────────────────── def route query(self, state workflowstate) > literal\["sql", "rag"] """route query to appropriate handler based on classification """ messages = state get("messages", \[]) if messages last = messages\[ 1] get("content", "") if '"sql"' in last lower() return "sql" return "rag" \# ───────────────────────────────────────────────────────────── \# helpers \# ───────────────────────────────────────────────────────────── def parse classification(self, response str) > dict\[str, any] import json 	 try 	 return json loads(response) 	 except json jsondecodeerror 	 return {"type" "rag", "confidence" 0 0} step 3 create the entrypoint create main py """application entrypoint for workflow agent """ from dotenv import load dotenv load dotenv(override=false) import logging from bridge agent sdk import run agent from src agents orchestrator import orchestratorworkflow logging basicconfig( level=logging info, format="%(asctime)s %(name)s %(levelname)s %(message)s" ) workflows = { "orchestrator" orchestratorworkflow, } agent name map = { "orchestrator" "orchestrator", } if name == " main " run agent( workflows, agent name map=agent name map, description="multi agent query orchestrator", ) how execution works when you call await workflow\ execute(agent input), the base class calls define state() to get the typeddict schema creates a stategraph and calls setup nodes() + setup edges() compiles the graph calls prepare initial state(agent input) to build the initial state dict creates the llm client and assigns it to self llm (via self create llm(state)) now that platform context is available activates automatic cadf audit hooks (if enable audit=true, the default) invokes the compiled graph with langfuse callbacks (if enable langfuse=true, the default) calls extract output(final state) to build the return value deactivates audit hooks in the finally block prepare initial state() — default state keys the sdk's default prepare initial state() auto decomposes executioncontext into \ state key \ \ source \ \ description \ input agent input content primary input content metadata agent input metadata execution metadata (thread id, etc ) context agent input context additional context previous outputs agent input previous outputs outputs from previous agents platform context extracted from executioncontext auth, account id, workflow id, etc agent payload same as platform context deprecated alias raw input extracted from executioncontext content, input, parameters, runtime data connection details from platform context list of connection dicts for external services override prepare initial state() to add custom keys while keeping the sdk's auto decomposition always call super() to get the base state \# agentbaseworkflow is your project's base class from base py (extends bridgebaseworkflowagent) class myworkflow(agentbaseworkflow) def prepare initial state(self, agent input agentinput) > dict\[str, any] 	"""extend the sdk's default state with custom keys """ base state = super() prepare initial state(agent input) \# base state already has platform context, raw input, connection details, etc base state update({ "status" workflowstatus pending, 	 "messages" \[], }) return base state def extract output(self, final state dict\[str, any]) > any """customize how final state maps to output """ return { 	 "content" final state get("output", ""), "status" final state get("status", workflowstatus error), } do not build platform context manually the sdk's super() prepare initial state() extracts it from executioncontext and validates all required fields if you skip super(), auth will fail in bridge dev and production modes advanced patterns retry loops def setup edges(self, graph stategraph) graph set entry point("attempt") graph add conditional edges( "attempt", self check result, { "success" "finalize", "retry" "attempt", "fail" end, } ) graph add edge("finalize", end) def check result(self, state) > str if state get("success") 	 return "success" elif state get("retry count", 0) < 3 return "retry" else 	 return "fail" sub workflow orchestration (pipeline pattern) use run sub workflow sync to call sub workflows from langgraph nodes (which are synchronous) create sub workflow instances once in init , not per invocation from bridge agent sdk import run sub workflow sync, agentinput class pipelineworkflow(bridgebaseworkflowagent) def init (self) 	super() init () 	self debug workflow = debugworkflow() self remediation workflow = remediationworkflow() 	 def node run debug(self, state) 	agent input = agentinput( content="debug task", 	 metadata={ 	 "thread id" state get("metadata", {}) get("thread id", "sub debug"), 	 "execution id" state get("metadata", {}) get("execution id", "unknown"), }, context={ "platform context" state\["platform context"], 	 "raw input" state\["raw input"], }, ) result = run sub workflow sync(self debug workflow, agent input) return { "debug report" result get("content", ""), "debug status" result get("status", "unknown"), } workflowconfig reference key workflowconfig fields and their defaults \ field \ \ type \ \ default \ \ description \ name str required workflow name version str "1 0 0" workflow version description str required workflow description llm provider str os getenv("llm provider", "openai") llm backend openai, azure openai, hosted llm, anthropic llm model str os getenv("llm model", "gpt 4 1 mini") model name (reads from env at class load) llm temperature float 0 7 sampling temperature max iterations int 20 max workflow iterations recursion limit int 200 max langgraph recursion depth enable langfuse bool true enable langfuse tracing for all llm calls enable audit bool true enable automatic cadf audit hooks (start/stop events) enable tenant db bool false enable multi tenant database isolation account code str none tenant code for db isolation (required when enable tenant db=true) automatic audit hooks (v1 0 16+) when enable audit=true (the default), the sdk automatically wraps execute() with cadf audit events you do not need to manually call post audit() for start/stop — the sdk emits start event when execute() begins sdk level events on llm calls, auth operations, connection lookups stop event when execute() completes (or fails) to opt out, set enable audit=false in workflowconfig guardrailsservice (v1 0 16+) the sdk includes guardrailsservice for pii/dlp redaction in audit logs and langfuse traces it is initialized automatically on bridgebaseworkflowagent when the guardrails endpoint is configured guardrails endpoint=https //your bridge host/kaif/v2/guardrails guardrails pii provider=azure # or "google" guardrails default config name=genaiassist default the guardrails attribute is automatically passed to auditservice and langfuse callbacks to redact pii from logged data tenant database (v1 0 16+) for agents that need per account data storage with schema isolation config = workflowconfig( name="my agent", description=" ", enable tenant db=true, account code="acme", ) \# in workflow nodes, access via self db async def store results node(self, state) await self db insert("execution log", {"status" "completed", "result" " "}) 	 rows = await self db select("execution log", where={"status" "completed"}) return {"stored" true} requires account postgres environment variables see tenantdatabasemanager docs for schema isolation, rls, and restricted roles testing workflows """tests for orchestratorworkflow """ import pytest from unittest mock import patch, magicmock from bridge agent sdk import agentinput from src agents orchestrator import orchestratorworkflow from src agents state import workflowstatus @pytest fixture def workflow() """create workflow instance """ return orchestratorworkflow() def test classifier routes correctly(workflow) """test that classifier produces expected state updates """ state = { "raw input" {"query" "show me incidents from last week"}, "platform context" {}, "status" workflowstatus pending, "messages" \[], } mock response = magicmock() mock response content = '{"type" "sql", "confidence" 0 95}' \# self llm is set by execute() at runtime; in unit tests, set it directly workflow\ llm = magicmock() workflow\ llm invoke return value = mock response result = workflow\ classifier node(state) assert result\["status"] == workflowstatus in progress assert len(result\["messages"]) == 1 @pytest mark asyncio async def test full workflow execution(workflow) """test complete workflow execution via execute() """ agent input = agentinput( content="what is our sla policy?", metadata={"thread id" "test 456"}, context={"agent payload" {"account id" "test account"}}, ) mock response = magicmock() mock response content = '{"type" "rag", "confidence" 0 9}' with patch object(workflow, ' create llm') as mock create llm mock llm = magicmock() mock llm invoke return value = mock response mock create llm return value = mock llm result = await workflow\ execute(agent input) assert result is not none