Skip to content

Instantly share code, notes, and snippets.

@kausmeows

kausmeows/workflow_with_bg_hook.py

Created January 9, 2026 05:01
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save kausmeows/1950fedb7a156309ef28552ece71eb97 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save kausmeows/1950fedb7a156309ef28552ece71eb97 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
workflow_with_bg_hook.py
"""
Example: Workflow with Conversational Intake and Background Research
This demonstrates a Workflow where:
1. An intake agent step has a conversation with the user
2. Once the user's name is captured, background research kicks off
3. The research results are stored in the workflow's shared session_state
4. Subsequent workflow steps/turns can access the research
Run with: python libs/agno/agno/test.py
"""
import asyncio
from typing import Optional
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.hooks.decorator import hook
from agno.models.openai import OpenAIChat
from agno.run import RunContext
from agno.run.agent import RunOutput
from agno.workflow import Workflow
from agno.workflow.step import Step
# =============================================================================
# Database for Persistence
# =============================================================================
db = SqliteDb(db_file="tmp/intake_workflow.db")
# =============================================================================
# Simulated Background Research Function
# =============================================================================
async def do_background_research(user_name: str) -> dict:
"""
Simulates background research on a user.
In production, this could call APIs, search databases, scrape LinkedIn, etc.
"""
print(f"\n[Background Task] Starting research on '{user_name}'...")
# Simulate async work (API calls, web scraping, etc.)
await asyncio.sleep(3)
# Return mock research results
research_results = {
"name": user_name,
"likely_interests": ["AI", "automation", "productivity"],
"suggested_topics": ["workflow optimization", "agent-based systems"],
"notes": f"This appears to be a tech-savvy user interested in AI tools.",
}
print(f"[Background Task] Research complete for '{user_name}'")
return research_results
# =============================================================================
# Background Hook - Runs After Agent Response (Non-blocking)
# =============================================================================
@hook(run_in_background=True)
async def background_research_hook(
run_output: RunOutput,
agent: Agent,
session_state: dict,
) -> None:
"""
Kicks off after intake agent responds - completely non-blocking.
The user gets their response immediately while this runs in the background.
Note: session_state here is the WORKFLOW's session_state (shared across all steps).
"""
user_name = session_state.get("user_name")
# Only start research if we have a name and haven't started yet
if user_name and not session_state.get("research_started"):
session_state["research_started"] = True
try:
# Do the actual research
research = await do_background_research(user_name)
# Update workflow session state with results
session_state["user_research"] = research
session_state["research_complete"] = True
print("[Background Task] Workflow session state updated with research results")
except Exception as e:
print(f"[Background Task] Research failed: {e}")
session_state["research_error"] = str(e)
# =============================================================================
# Tool for Agent to Capture User Information (updates workflow session_state)
# =============================================================================
def capture_user_info(
run_context: RunContext,
name: str,
company: Optional[str] = None,
) -> str:
"""
Capture user information during intake conversation.
This updates the workflow's shared session_state.
Args:
name: The user's name
company: The user's company (optional)
Returns:
Confirmation message
"""
if run_context.session_state is None:
run_context.session_state = {}
# These updates go to the workflow's session_state
run_context.session_state["user_name"] = name
if company:
run_context.session_state["user_company"] = company
return f"Got it, {name}! I'm looking forward to helping you today."
# =============================================================================
# Intake Agent - First step in the workflow
# =============================================================================
intake_agent = Agent(
name="Intake Agent",
model=OpenAIChat(id="gpt-4o-mini"),
tools=[capture_user_info],
post_hooks=[background_research_hook],
instructions="""\
You are a friendly intake assistant having a conversation with a new user.
Your goals:
1. Greet the user warmly
2. Learn their name (use the capture_user_info tool when they tell you)
3. Understand what they need help with
4. Provide helpful, personalized responses
Current session context:
- User name: {user_name}
- Research complete: {research_complete}
- Research findings: {user_research}
Guidelines:
- Be conversational and friendly
- When the user tells you their name, ALWAYS use capture_user_info to record it
- If research is available, naturally incorporate relevant insights into your responses
- Don't mention "research" explicitly - just use the information naturally
""",
markdown=True,
)
# =============================================================================
# Advisor Agent - Second step that can use research results
# =============================================================================
advisor_agent = Agent(
name="AI Advisor",
model=OpenAIChat(id="gpt-4o-mini"),
instructions="""\
You are an AI solutions advisor providing personalized recommendations.
User context:
- User name: {user_name}
- User company: {user_company}
- Research findings: {user_research}
Use the research findings to provide tailored, relevant recommendations.
Be specific and reference the user's likely interests when making suggestions.
""",
markdown=True,
)
# =============================================================================
# Workflow Definition with Shared Session State
# =============================================================================
intake_workflow = Workflow(
name="Intake and Advisory Workflow",
description="A workflow that handles user intake and provides personalized AI recommendations",
steps=[
Step(name="intake", agent=intake_agent, description="Handle user intake and gather information"),
Step(name="advise", agent=advisor_agent, description="Provide personalized AI recommendations"),
],
# Shared session_state across all steps
session_state={
"user_name": None,
"user_company": None,
"user_research": None,
"research_started": False,
"research_complete": False,
},
db=db,
)
# =============================================================================
# Main Demo
# =============================================================================
async def main():
"""Run the intake workflow demo."""
print("=" * 70)
print("Workflow with Conversational Intake and Background Research Demo")
print("=" * 70)
print()
# Run 1: Initial greeting - goes through intake step
print("Run 1: User says hello (intake step)")
print("-" * 50)
await intake_workflow.aprint_response(
"Hi there! I'm looking for help with building AI agents.",
stream=True,
)
print(f"\nWorkflow session state: {intake_workflow.get_session_state()}")
print()
# Run 2: User provides name - triggers background research
print("Run 2: User provides name (triggers background research)")
print("-" * 50)
await intake_workflow.aprint_response(
"My name is Alex Chen and I work at TechCorp.",
stream=True,
)
print(f"\nWorkflow session state: {intake_workflow.get_session_state()}")
print()
# Wait for background research to complete
print("Waiting for background research to complete...")
await asyncio.sleep(4)
print()
# Run 3: Ask for recommendations - research should be available now
print("Run 3: Ask for recommendations (research should be available)")
print("-" * 50)
await intake_workflow.aprint_response(
"What kind of AI solutions would you recommend for my use case?",
stream=True,
)
print(f"\nFinal workflow session state: {intake_workflow.get_session_state()}")
if __name__ == "__main__":
asyncio.run(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment