Copy
Ask AI
"""
History In Function
===================
Demonstrates reading workflow history inside a custom function step for strategic content planning.
"""
import json
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIChat
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow
# ---------------------------------------------------------------------------
# Define Analysis Function
# ---------------------------------------------------------------------------
def analyze_content_strategy(step_input: StepInput) -> StepOutput:
current_topic = step_input.input or ""
research_data = step_input.get_last_step_content() or ""
history_data = step_input.get_workflow_history(num_runs=5)
def extract_keywords(text: str) -> set:
stop_words = {
"create",
"content",
"about",
"write",
"the",
"a",
"an",
"how",
"is",
"of",
"this",
"that",
"in",
"on",
"for",
"to",
}
words = set(text.lower().split()) - stop_words
keyword_map = {
"ai": ["ai", "artificial", "intelligence"],
"ml": ["machine", "learning", "ml"],
"healthcare": ["medical", "health", "healthcare", "medicine"],
"blockchain": ["crypto", "cryptocurrency", "blockchain"],
}
expanded_keywords = set(words)
for word in list(words):
for synonyms in keyword_map.values():
if word in synonyms:
expanded_keywords.update([word])
return expanded_keywords
current_keywords = extract_keywords(current_topic)
max_possible_overlap = len(current_keywords)
topic_overlaps = []
covered_topics = []
for input_request, _content_output in history_data:
if input_request:
covered_topics.append(input_request.lower())
previous_keywords = extract_keywords(input_request)
overlap = len(current_keywords.intersection(previous_keywords))
if overlap > 0:
topic_overlaps.append(overlap)
topic_overlap = max(topic_overlaps) if topic_overlaps else 0
overlap_percentage = (topic_overlap / max(max_possible_overlap, 1)) * 100
diversity_score = len(set(covered_topics)) / max(len(covered_topics), 1)
recommendations = []
if overlap_percentage > 60:
recommendations.append(
"HIGH OVERLAP detected - consider a fresh angle or advanced perspective"
)
elif overlap_percentage > 30:
recommendations.append(
"MODERATE OVERLAP detected - differentiate your approach"
)
if diversity_score < 0.6:
recommendations.append(
"Low content diversity - explore different aspects of the topic"
)
if len(history_data) > 0:
recommendations.append(
f"Building on {len(history_data)} previous content pieces - ensure progression"
)
strategy_analysis = {
"content_topic": current_topic,
"historical_coverage": {
"previous_topics": covered_topics[-3:],
"topic_overlap_score": topic_overlap,
"overlap_percentage": round(overlap_percentage, 1),
"content_diversity": diversity_score,
},
"strategic_recommendations": recommendations,
"research_summary": research_data[:500] + "..."
if len(research_data) > 500
else research_data,
"suggested_angle": "unique perspective"
if overlap_percentage > 30
else "comprehensive overview",
"content_gap_analysis": {
"avoid_repeating": [
topic
for topic in covered_topics
if any(word in current_topic.lower() for word in topic.split()[:2])
],
"build_upon": "previous insights"
if len(history_data) > 0
else "foundational knowledge",
},
}
formatted_analysis = f"""
CONTENT STRATEGY ANALYSIS
========================
STRATEGIC OVERVIEW:
- Topic: {strategy_analysis["content_topic"]}
- Previous Content Count: {len(history_data)}
- Keyword Overlap: {strategy_analysis["historical_coverage"]["topic_overlap_score"]} keywords ({strategy_analysis["historical_coverage"]["overlap_percentage"]}%)
- Content Diversity: {strategy_analysis["historical_coverage"]["content_diversity"]:.2f}
RECOMMENDATIONS:
{chr(10).join([f"- {rec}" for rec in strategy_analysis["strategic_recommendations"]])}
RESEARCH FOUNDATION:
{strategy_analysis["research_summary"]}
CONTENT POSITIONING:
- Suggested Angle: {strategy_analysis["suggested_angle"]}
- Build Upon: {strategy_analysis["content_gap_analysis"]["build_upon"]}
- Differentiate From: {", ".join(strategy_analysis["content_gap_analysis"]["avoid_repeating"]) if strategy_analysis["content_gap_analysis"]["avoid_repeating"] else "No similar content found"}
CREATIVE DIRECTION:
Based on historical analysis, focus on providing {strategy_analysis["suggested_angle"]} while ensuring the content complements rather than duplicates previous work.
STRUCTURED_DATA: {json.dumps(strategy_analysis, indent=2)}
"""
return StepOutput(content=formatted_analysis.strip())
# ---------------------------------------------------------------------------
# Create Workflow
# ---------------------------------------------------------------------------
def create_content_workflow() -> Workflow:
research_step = Step(
name="Content Research",
agent=Agent(
name="Research Specialist",
model=OpenAIChat(id="gpt-4o"),
instructions=[
"You are an expert research specialist for content creation.",
"Conduct thorough research on the requested topic.",
"Gather current trends, key insights, statistics, and expert perspectives.",
"Structure your research with clear sections: Overview, Key Points, Recent Developments, Expert Insights.",
"Prioritize accurate, up-to-date information from credible sources.",
"Keep research comprehensive but concise for content creators to use.",
],
),
)
strategy_step = Step(
name="Content Strategy Analysis",
executor=analyze_content_strategy,
description="Analyze content strategy using historical data to prevent duplication and identify opportunities",
)
writer_step = Step(
name="Strategic Content Creation",
agent=Agent(
name="Content Strategist",
model=OpenAIChat(id="gpt-4o"),
instructions=[
"You are a strategic content writer who creates high-quality, unique content.",
"Use the research and strategic analysis to create compelling content.",
"Follow the strategic recommendations to ensure content uniqueness.",
"Structure content with: Hook, Main Content, Key Takeaways, Call-to-Action.",
"Ensure your content builds upon previous work rather than repeating it.",
"Include 'Target Audience:' and 'Content Type:' at the end for tracking.",
"Make content engaging, actionable, and valuable to readers.",
],
),
)
return Workflow(
name="Strategic Content Creation",
description="Research -> Strategic Analysis -> Content Creation with historical awareness",
db=SqliteDb(db_file="tmp/content_workflow.db"),
steps=[research_step, strategy_step, writer_step],
add_workflow_history_to_steps=True,
)
# ---------------------------------------------------------------------------
# Run Workflow
# ---------------------------------------------------------------------------
def demo_content_workflow() -> None:
workflow = create_content_workflow()
print("Strategic Content Creation Workflow")
print("Flow: Research -> Strategy Analysis -> Content Writing")
print("")
print("This workflow prevents duplicate content and ensures strategic progression")
print("")
print("Try these content requests:")
print("- 'Create content about AI in healthcare'")
print("- 'Write about machine learning applications' (will detect overlap)")
print("- 'Content on blockchain technology' (different topic)")
print("")
print("Type 'exit' to quit")
print("-" * 70)
workflow.cli_app(
session_id="content_strategy_demo",
user="Content Manager",
emoji="",
stream=True,
)
if __name__ == "__main__":
demo_content_workflow()
Run the Example
Copy
Ask AI
# Clone and setup repo
git clone https://github.com/agno-agi/agno.git
cd agno/cookbook/04_workflows/06_advanced_concepts/history
# Create and activate virtual environment
./scripts/demo_setup.sh
source .venvs/demo/bin/activate
python history_in_function.py