Practical agentic AI workflow patterns for product managers building LLM systems. Real patterns from production — not tutorials.
Built by Manvendra Kumar · Senior AI Product Manager
I've shipped LLM systems at Redo (claims automation) and CareBow (healthcare AI). These are the agentic patterns that actually worked in production — along with the PM context for when to use each one.
This isn't a LangChain tutorial. It's a pattern library with:
- What the pattern does
- When to use it (and when not to)
- A minimal implementation sketch
- The PM decision that justified it
What it does: Each LLM call feeds directly into the next. Output of step N is input to step N+1.
When to use:
- You have a well-defined, ordered process
- Each step has a clear input/output contract
- Low ambiguity in the flow
When NOT to use:
- The process is branching or conditional
- You need the agent to decide what to do next
Real use: CareBow intake flow — Symptom Collection → Severity Scoring → Caregiver Match → Care Plan Draft
from langchain.chains import LLMChain, SequentialChain
from langchain.prompts import PromptTemplate
# Step 1: Extract structured symptoms
symptom_prompt = PromptTemplate(
input_variables=["raw_intake"],
template="""Extract key symptoms from this intake form. Return JSON.
Intake: {raw_intake}
Output: {{"symptoms": [], "severity": "", "urgency": ""}}"""
)
# Step 2: Score severity
severity_prompt = PromptTemplate(
input_variables=["symptoms"],
template="""Given these symptoms: {symptoms}
Score severity 1-10 and recommend care level: home, urgent, ER."""
)
chain = SequentialChain(
chains=[symptom_chain, severity_chain],
input_variables=["raw_intake"],
output_variables=["care_recommendation"]
)PM decision that justified this: We knew the flow wouldn't change for MVP. Sequential chain gave us a clear audit trail for each step — critical for HIPAA compliance review.
What it does: A classifier LLM decides which specialized chain to call based on input.
When to use:
- Multiple distinct cases that need different handling
- You want to route to specialized prompts per case type
- You need to avoid one giant prompt trying to handle everything
Real use: Redo claims routing — Classify claim type → dispatch to Standard, Fraud, High-Value, or Manual-Review chain
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain, RouterOutputParser
# Define destination chains
destination_chains = {
"standard_claim": standard_claim_chain,
"fraud_signal": fraud_review_chain,
"high_value": high_value_chain,
"manual": manual_review_chain,
}
# Router prompt
router_template = """Given this claim, route it to the right handler.
Claim: {input}
Options:
- standard_claim: routine return, clear documentation
- fraud_signal: inconsistencies, repeated patterns, suspicious timing
- high_value: claim amount > $500
- manual: ambiguous, missing data, or novel case type
Route to:"""
router_chain = LLMRouterChain.from_llm(llm, router_prompt)
multi_chain = MultiPromptChain(
router_chain=router_chain,
destination_chains=destination_chains,
default_chain=manual_review_chain
)PM decision that justified this: One prompt was classifying AND processing. Accuracy was 71%. Splitting router from processor got us to 88% before we added few-shot examples.
What it does: LLM reasons, selects a tool, observes the result, reasons again. Continues until it has enough to answer.
When to use:
- The agent needs to look something up before answering
- Multiple tool calls may be needed in unpredictable order
- You're building something that behaves like a smart assistant
When NOT to use:
- You need deterministic, auditable output (use sequential chain instead)
- Latency is critical — ReAct loops add round-trips
- You're in a regulated context without explainability
from langchain.agents import initialize_agent, AgentType
from langchain.tools import Tool
# Define tools the agent can use
tools = [
Tool(
name="search_carrier_policy",
func=search_carrier_db,
description="Look up carrier return policy by carrier_id and product_category"
),
Tool(
name="check_claim_history",
func=check_claim_history,
description="Check if customer has prior claims in the last 90 days"
),
Tool(
name="calculate_refund",
func=calculate_refund_amount,
description="Calculate refund amount given claim details and policy"
)
]
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
verbose=True
)
result = agent.run("Process this return claim: [claim details]")PM decision that justified this: Claims require carrier policy lookup before any decision. Static prompts couldn't handle 200+ carrier variations. Agent with tool access replaced a 200-line lookup table.
What it does: Agent pauses and requests human confirmation before taking an irreversible action.
When to use:
- Output has irreversible consequences (issuing refunds, sending communications)
- Confidence is below threshold
- Regulatory context requires human approval on record
Real use: Redo — claims above $500 always pause for human review regardless of LLM confidence
from langchain.callbacks import HumanApprovalCallbackHandler
def should_check(serialized_obj: dict) -> bool:
"""Require human approval for high-value or low-confidence actions."""
if serialized_obj.get("action") == "issue_refund":
if serialized_obj.get("amount", 0) > 500:
return True
if serialized_obj.get("confidence", 1.0) < 0.85:
return True
return False
callbacks = [HumanApprovalCallbackHandler(should_check=should_check)]
agent = initialize_agent(
tools=tools,
llm=llm,
agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
callbacks=callbacks
)HITL Decision Matrix:
| Condition | Action |
|---|---|
| Confidence ≥ 85%, amount < $500 | Auto-approve |
| Confidence 60–85% | Flag for review |
| Confidence < 60% | Always human |
| Amount > $500, confidence > 90% | Auto-approve with dual log |
| Amount > $500, confidence ≤ 90% | Mandatory human |
| Fraud signal present | Always human |
PM decision that justified this: Before HITL, 6% of auto-approved claims were incorrect — manageable on volume but not at scale. HITL on low-confidence + high-value reduced error rate to 0.4%.
What it does: Retrieves relevant documents from a vector store before generating. Grounds the LLM in your actual data.
When to use:
- LLM needs knowledge that wasn't in its training data
- You have proprietary docs, policies, or knowledge base
- You need citations or traceability
Real use: Mopshy AI — SMB clients' internal SOPs and product FAQs retrieved before answering customer queries
from langchain.vectorstores import Pinecone
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
# Set up vector store
embeddings = OpenAIEmbeddings()
vectorstore = Pinecone.from_documents(
documents=company_docs,
embedding=embeddings,
index_name="company-knowledge-base"
)
# Create RAG chain
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
return_source_documents=True
)
result = qa_chain({"query": "What's the return window for electronics?"})
# Returns answer + source documents for traceabilityPM decision that justified this: Generic LLM hallucinated carrier policies. RAG grounded answers in actual carrier contracts. False positive rate on policy questions dropped from 22% to 3%.
What it does: Multiple specialized agents hand off to each other. Each agent owns one part of the problem.
When to use:
- Problem is too complex for one agent
- Different parts need different tools or expertise
- You want independent error handling per stage
Real use: Mopshy AI sales pipeline — Prospecting Agent → Qualification Agent → Outreach Agent → CRM Update Agent
from langchain.agents import AgentExecutor
class MultiAgentPipeline:
def __init__(self):
self.prospecting_agent = AgentExecutor(...) # finds leads
self.qualification_agent = AgentExecutor(...) # scores fit
self.outreach_agent = AgentExecutor(...) # drafts messages
self.crm_agent = AgentExecutor(...) # logs to CRM
def run(self, company_target: str) -> dict:
# Stage 1: Find prospects
prospects = self.prospecting_agent.run(
f"Find 10 decision-makers at {company_target}"
)
# Stage 2: Qualify each prospect
qualified = self.qualification_agent.run(
f"Score fit for these prospects: {prospects}"
)
# Stage 3: Draft outreach for top 3
outreach = self.outreach_agent.run(
f"Draft personalized outreach for: {qualified[:3]}"
)
# Stage 4: Log everything
self.crm_agent.run(f"Log to CRM: {outreach}")
return {"prospects": qualified, "outreach": outreach}PM decision that justified this: Single agent was hitting context limits on large prospecting runs and hallucinating CRM fields. Splitting by concern let us tune each agent independently and catch failures at the handoff point.
| Situation | Pattern |
|---|---|
| Linear, predictable process | Sequential Chain |
| Multiple case types to handle differently | Router Chain |
| Needs to look things up dynamically | Agent with Tools |
| Irreversible actions or compliance context | HITL Agent |
| LLM needs your proprietary knowledge | RAG |
| Problem is too big for one agent | Multi-Agent |
Built by Manvendra Kumar — Senior AI Product Manager Open to Senior PM roles at AI-native companies · manvendrakumar.com