arindam200--awesome-ai-apps
wehub-resource-sync
630 行
22 KiB
Python
630 行
22 KiB
Python
"""Chat parsing and follow-up helpers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from models import AgentRunResult, CompanyContext
|
|
|
|
|
|
FIELD_ALIASES = {
|
|
"company": "company_name",
|
|
"company name": "company_name",
|
|
"product": "product",
|
|
"category": "product",
|
|
"product/category": "product",
|
|
"audience": "audience",
|
|
"target audience": "audience",
|
|
"keywords": "seed_keywords",
|
|
"seed keywords": "seed_keywords",
|
|
"competitors": "competitors",
|
|
"existing topics": "existing_topics",
|
|
"existing content": "existing_topics",
|
|
}
|
|
|
|
|
|
def split_csv(value: str) -> list[str]:
|
|
return [item.strip() for item in value.split(",") if item.strip()]
|
|
|
|
|
|
def apply_prompt_context(prompt: str, context: dict) -> bool:
|
|
"""Parse structured or natural chat context into the mutable Streamlit context dict."""
|
|
changed = False
|
|
for raw_line in prompt.replace(";", "\n").splitlines():
|
|
if ":" not in raw_line:
|
|
continue
|
|
key, value = raw_line.split(":", 1)
|
|
field = FIELD_ALIASES.get(key.strip().lower())
|
|
if not field or not value.strip():
|
|
continue
|
|
if field in {"seed_keywords", "competitors", "existing_topics"}:
|
|
context[field] = split_csv(value)
|
|
else:
|
|
context[field] = value.strip()
|
|
changed = True
|
|
changed = _apply_natural_context(prompt, context) or changed
|
|
return changed
|
|
|
|
|
|
def _apply_natural_context(prompt: str, context: dict) -> bool:
|
|
changed = False
|
|
normalized = " ".join(prompt.split())
|
|
|
|
if not context["company_name"].strip():
|
|
company = _first_match(
|
|
normalized,
|
|
[
|
|
r"\b(?:built|created|launched)\s+([A-Za-z0-9-]+\.[A-Za-z]{2,})(?:,|\.)",
|
|
r"\b(?:I\s+)?(?:run|own|manage|founded?)\s+([A-Za-z0-9-]+\.[A-Za-z]{2,})(?:,|\.|\s|$)",
|
|
r"\b(?:for|about|around)\s+([A-Za-z0-9-]+\.[A-Za-z]{2,})(?:,|\.|\s|$)",
|
|
r"(?:devrel|developer relations)\s+for\s+([A-Za-z][A-Za-z0-9 .&-]{1,40}?)(?:,|\.|\s+a\s)",
|
|
r"\b(?:built|created|launched)\s+([A-Za-z][A-Za-z0-9 .&-]{1,40}?)(?:,|\.)",
|
|
r"(?:leading|run|own|manage)\s+content\s+for\s+([A-Za-z][A-Za-z0-9 .&-]{1,40}?)(?:,|\sand\s|\.)",
|
|
r"(?:company|startup|team)\s+(?:is|called)\s+([A-Za-z][A-Za-z0-9 .&-]{1,40}?)(?:,|\.|\sand\s)",
|
|
r"\bfor\s+([A-Z][A-Za-z0-9 .&-]{1,40}?)(?:,?\s+and\s+I\s+want|,?\s+we\s+want)",
|
|
],
|
|
)
|
|
if company:
|
|
context["company_name"] = _clean_phrase(company)
|
|
changed = True
|
|
|
|
if not context["product"].strip():
|
|
product = _first_match(
|
|
normalized,
|
|
[
|
|
r"(?:built|created|launched|run|own|manage|founded?)\s+[A-Za-z][A-Za-z0-9 .&-]{1,40}?,\s+(?:a|an|the)\s+(.+?)(?:\.|\s+for\s+)",
|
|
r"(?:devrel|developer relations)\s+for\s+[A-Za-z][A-Za-z0-9 .&-]{1,40}?,\s+(?:a|an|the)\s+(.+?)(?:\.|\s+Our\s+audience\s+)",
|
|
r"opportunities\s+for\s+our\s+(.+?)(?:\.|\s+Our\s+audience\s+is)",
|
|
r"(?:product|category)\s+(?:is|=)\s+(.+?)(?:\.|,|\s+and\s)",
|
|
r"(?:it'?s\s+a|its\s+a|it\s+is\s+a)\s+(.+?(?:tool|platform|framework|sdk|api|system|service|app))(?:\.|,|\s|$)",
|
|
r"building\s+(?:a|an|the|our)?\s*(.+?(?:framework|platform|tool|sdk|api|system|product))(?:\.|,|\s+for\s)",
|
|
],
|
|
)
|
|
if product:
|
|
context["product"] = _clean_phrase(product)
|
|
changed = True
|
|
|
|
if not context["audience"].strip():
|
|
audience = _first_match(
|
|
normalized,
|
|
[
|
|
r"(?:my|our)\s+audience\s+is\s+(.+?)(?:\.|\s+They\s+care\s+about)",
|
|
r"(?:our\s+)?audience\s+is\s+(.+?)(?:\.|\s+They\s+care\s+about)",
|
|
r"\bhelp\s+(.+?)\s+understand\b",
|
|
r"(?:for|targeting)\s+((?:senior\s+)?(?:AI|ML|software|platform|engineering|backend|frontend|devops)[^.]{10,140}?)(?:\.|\s+who\s+care)",
|
|
],
|
|
)
|
|
if audience:
|
|
context["audience"] = _clean_phrase(audience)
|
|
changed = True
|
|
|
|
if not context["competitors"]:
|
|
competitors = _first_match(
|
|
normalized,
|
|
[
|
|
r"(?:competitors|adjacent tools|alternatives)[^.]{0,40}\s+include\s+(.+?)(?:\.|$)",
|
|
r"(?:against|versus|vs\.?)\s+(.+?)(?:\.|$)",
|
|
],
|
|
)
|
|
if competitors:
|
|
context["competitors"] = _split_natural_list(competitors)
|
|
changed = bool(context["competitors"]) or changed
|
|
|
|
if not context["seed_keywords"]:
|
|
keyword_chunks: list[str] = []
|
|
cares = _first_match(normalized, [r"care\s+about\s+(.+?)(?:\.|$)"])
|
|
interested = _first_match(normalized, [r"interested\s+in\s+topics\s+that\s+could\s+become\s+(.+?)(?:\.|$)"])
|
|
if cares:
|
|
keyword_chunks.extend(_split_natural_list(cares))
|
|
if interested:
|
|
keyword_chunks.extend(_split_natural_list(interested))
|
|
around = _first_match(normalized, [r"\baround\s+(.+?)(?:\.|$)"])
|
|
if around:
|
|
keyword_chunks.extend(_split_natural_list(around))
|
|
if context["product"]:
|
|
keyword_chunks.extend(_product_keywords(context["product"]))
|
|
context["seed_keywords"] = list(dict.fromkeys(item for item in keyword_chunks if len(item) > 2))[:10]
|
|
changed = bool(context["seed_keywords"]) or changed
|
|
|
|
return changed
|
|
|
|
|
|
def _first_match(text: str, patterns: list[str]) -> str | None:
|
|
for pattern in patterns:
|
|
match = re.search(pattern, text, flags=re.IGNORECASE)
|
|
if match:
|
|
return match.group(1).strip()
|
|
return None
|
|
|
|
|
|
def _clean_phrase(value: str) -> str:
|
|
value = re.sub(r"\s+", " ", value).strip(" .,")
|
|
value = re.sub(r"^(?:a|an|the|our)\s+", "", value, flags=re.IGNORECASE)
|
|
return value
|
|
|
|
|
|
def _split_natural_list(value: str) -> list[str]:
|
|
value = value.replace(" and ", ", ")
|
|
value = value.replace(" or ", ", ")
|
|
return [
|
|
_clean_phrase(item)
|
|
for item in value.split(",")
|
|
if _clean_phrase(item)
|
|
]
|
|
|
|
|
|
def _product_keywords(product: str) -> list[str]:
|
|
words = [word.strip(".,").lower() for word in product.split()]
|
|
pairs = [" ".join(words[index : index + 2]) for index in range(max(0, len(words) - 1))]
|
|
return [_clean_phrase(product), *pairs]
|
|
|
|
|
|
def missing_fields(context: dict) -> list[str]:
|
|
missing = []
|
|
if not context["company_name"].strip() and not context["product"].strip():
|
|
missing.append("company or product/category")
|
|
elif not context["product"].strip():
|
|
context["product"] = context["company_name"].strip()
|
|
if not context["audience"].strip():
|
|
missing.append("target audience")
|
|
return missing
|
|
|
|
|
|
def has_required_context(context: dict) -> bool:
|
|
return not missing_fields(context)
|
|
|
|
|
|
def wants_brief_run(prompt: str, context: dict | None = None) -> bool:
|
|
lowered = prompt.lower()
|
|
if wants_explicit_research_run(prompt):
|
|
return True
|
|
if wants_memory_lookup(prompt) or wants_current_report_followup(prompt):
|
|
return False
|
|
|
|
run_phrases = (
|
|
"run the brief",
|
|
"run trend digest",
|
|
"run the trend digest",
|
|
"trend digest",
|
|
"talk ideas",
|
|
"blog ideas",
|
|
"devrel ideas",
|
|
"research trends",
|
|
"run gap analysis",
|
|
"generate the report",
|
|
"create the report",
|
|
"generate the brief",
|
|
"create the brief",
|
|
"start the brief",
|
|
"run research",
|
|
"start research",
|
|
"generate it",
|
|
"run it",
|
|
)
|
|
if any(phrase in lowered for phrase in run_phrases):
|
|
return True
|
|
|
|
action_terms = (
|
|
"find",
|
|
"search",
|
|
"research",
|
|
"look for",
|
|
"recommend",
|
|
"suggest",
|
|
"analyze",
|
|
"analyse",
|
|
"discover",
|
|
"identify",
|
|
"give me",
|
|
"show me",
|
|
"what should",
|
|
"help me",
|
|
)
|
|
publishing_terms = (
|
|
"content idea",
|
|
"content ideas",
|
|
"topic",
|
|
"topics",
|
|
"article",
|
|
"articles",
|
|
"blog",
|
|
"blogs",
|
|
"talk",
|
|
"talks",
|
|
"tutorial",
|
|
"tutorials",
|
|
"devrel",
|
|
"trend",
|
|
"trends",
|
|
"publish",
|
|
"publishing",
|
|
"opportunit",
|
|
"what to write",
|
|
"write about",
|
|
)
|
|
has_action = any(action in lowered for action in action_terms)
|
|
has_publishing_term = any(
|
|
term in lowered for term in publishing_terms
|
|
)
|
|
if has_action and has_publishing_term:
|
|
return True
|
|
|
|
generic_idea_terms = (
|
|
"idea",
|
|
"ideas",
|
|
"suggestion",
|
|
"suggestions",
|
|
"recommendation",
|
|
"recommendations",
|
|
)
|
|
generic_run_terms = (
|
|
"suggest some",
|
|
"suggest ideas",
|
|
"some ideas",
|
|
"content ideas",
|
|
"please search",
|
|
"search for me",
|
|
"research for me",
|
|
"look for ideas",
|
|
"find and analyze",
|
|
"find and analyse",
|
|
"analyze for me",
|
|
"analyse for me",
|
|
"do it",
|
|
"go ahead",
|
|
)
|
|
if context and has_required_context(context):
|
|
return any(term in lowered for term in generic_run_terms) or (
|
|
has_action and any(term in lowered for term in generic_idea_terms)
|
|
)
|
|
|
|
return False
|
|
|
|
|
|
def wants_explicit_research_run(prompt: str) -> bool:
|
|
lowered = prompt.lower()
|
|
old_session_terms = ("before", "past", "previous", "previously", "history", "memory")
|
|
if any(term in lowered for term in old_session_terms):
|
|
return False
|
|
|
|
explicit_research_terms = (
|
|
"research developer conversations",
|
|
"research what developers",
|
|
"research recent",
|
|
"research trends",
|
|
"run research",
|
|
"start research",
|
|
"check dev.to",
|
|
"dev.to saturation",
|
|
"article saturation",
|
|
"hacker news demand",
|
|
"hn demand",
|
|
"recommend technical blog",
|
|
"recommend technical blogs",
|
|
"recommend talk",
|
|
"recommend talks",
|
|
"publish next",
|
|
"what should we publish",
|
|
)
|
|
if any(term in lowered for term in explicit_research_terms):
|
|
return True
|
|
|
|
action_terms = ("find", "search", "research", "analyze", "analyse", "recommend", "suggest")
|
|
source_terms = ("hn", "hacker news", "dev.to", "devto", "developer conversations", "article saturation")
|
|
output_terms = ("ideas", "topics", "blog", "blogs", "talk", "talks", "trend", "trends", "publish")
|
|
return (
|
|
any(term in lowered for term in action_terms)
|
|
and any(term in lowered for term in source_terms)
|
|
and any(term in lowered for term in output_terms)
|
|
)
|
|
|
|
|
|
def wants_current_report_followup(prompt: str) -> bool:
|
|
lowered = prompt.lower()
|
|
old_session_terms = ("before", "past", "previous", "previously", "history", "memory")
|
|
if any(term in lowered for term in old_session_terms):
|
|
return False
|
|
|
|
report_terms = (
|
|
"latest",
|
|
"current",
|
|
"this report",
|
|
"the report",
|
|
"the digest",
|
|
"this digest",
|
|
"idea",
|
|
"ideas",
|
|
"topic",
|
|
"topics",
|
|
"trend",
|
|
"trends",
|
|
"evidence",
|
|
"source",
|
|
"sources",
|
|
"what did we find",
|
|
"what we found",
|
|
"what did you find",
|
|
"what did we get",
|
|
"what was suggested",
|
|
"what topics were suggested",
|
|
"what ideas were suggested",
|
|
"make these",
|
|
"make it",
|
|
"more technical",
|
|
"comparison",
|
|
"vs",
|
|
)
|
|
question_or_refinement = (
|
|
"?" in prompt
|
|
or any(term in lowered for term in ("show", "explain", "summarize", "recap", "make", "turn", "focus"))
|
|
)
|
|
return question_or_refinement and any(term in lowered for term in report_terms)
|
|
|
|
|
|
def render_context_summary(context: dict) -> str:
|
|
company_context = CompanyContext(**context)
|
|
lines = ["Got it. I'll treat this as the DevRel research context:"]
|
|
lines.append(f"- {company_context.company_name or 'Missing company'} is the company.")
|
|
lines.append(f"- The product/category is {company_context.product or 'missing'}." )
|
|
lines.append(f"- The audience is {company_context.audience or 'missing'}." )
|
|
if company_context.seed_keywords:
|
|
lines.append(f"- I’ll start from these signals: {', '.join(company_context.seed_keywords)}." )
|
|
else:
|
|
lines.append("- I’ll derive HN and DEV search angles from the product and audience.")
|
|
if company_context.competitors:
|
|
lines.append(f"- I’ll keep these alternatives in view: {', '.join(company_context.competitors)}.")
|
|
if company_context.existing_topics:
|
|
lines.append(f"- Existing topics to account for: {', '.join(company_context.existing_topics)}.")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def conversational_response(
|
|
prompt: str,
|
|
context: dict,
|
|
result: AgentRunResult | None,
|
|
) -> str | None:
|
|
lowered = prompt.lower().strip()
|
|
missing = missing_fields(context)
|
|
|
|
if any(term in lowered for term in ("what can you do", "how does this work", "what do you do")):
|
|
return (
|
|
"I help turn developer demand into a trend digest and DevRel talk/blog ideas. Once you give me "
|
|
"enough context, I run an Agno multi-agent team that checks HN demand, DEV.to supply, and "
|
|
"Engram memory in parallel. For normal follow-up questions like this, I won't call HN or DEV."
|
|
)
|
|
|
|
if any(term in lowered for term in ("do you call hn", "call dev", "hacker news every time", "dev every time")):
|
|
return (
|
|
"No. I only call Hacker News and DEV.to when you ask me to find, recommend, or analyze talk/blog ideas or trends. "
|
|
"Regular chat follow-ups use the current session context, the latest generated report, or Engram chat storage."
|
|
)
|
|
|
|
if any(term in lowered for term in ("what context", "what do you know", "current context", "what have you got")):
|
|
return render_context_summary(context)
|
|
|
|
summary_triggers = (
|
|
"summary", "recap", "what did we find", "what we found",
|
|
"what we researched", "what did we research", "what have we",
|
|
"what we got", "what did we get", "what did you find",
|
|
"what was researched", "researched about",
|
|
)
|
|
if result is not None and any(term in lowered for term in summary_triggers):
|
|
ideas = result.report.content_ideas[:3]
|
|
trends = result.report.trend_digest[:3]
|
|
if not ideas and not trends:
|
|
return result.report.summary
|
|
lines = [result.report.summary, ""]
|
|
if trends:
|
|
lines.append("Top trends from the latest report:")
|
|
lines.extend(
|
|
f"{index}. {trend.topic} - intensity {trend.intensity_score}"
|
|
for index, trend in enumerate(trends, 1)
|
|
)
|
|
lines.append("")
|
|
if ideas:
|
|
lines.append("Top talk/blog ideas from the latest report:")
|
|
lines.extend(
|
|
f"{index}. {idea.title} ({idea.format}) - score {idea.score}"
|
|
for index, idea in enumerate(ideas, 1)
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
if result is not None and any(term in lowered for term in ("why", "explain", "tell me more")):
|
|
topic_number = _first_number(lowered)
|
|
if topic_number is not None and 0 < topic_number <= len(result.report.content_ideas):
|
|
idea = result.report.content_ideas[topic_number - 1]
|
|
return (
|
|
f"Idea {topic_number}, **{idea.title}**, is interesting because {idea.angle}\n\n"
|
|
f"The gap I'd lean into: {idea.dev_gap}\n\n"
|
|
f"Format: {idea.format} | Confidence: {idea.confidence}"
|
|
)
|
|
|
|
if wants_brief_run(prompt, context) and missing:
|
|
return (
|
|
"I can do that, but I still need "
|
|
+ (missing[0] if len(missing) == 1 else ", ".join(missing[:-1]) + f", and {missing[-1]}")
|
|
+ ". You can say it naturally; no labels required."
|
|
)
|
|
|
|
if "?" in prompt:
|
|
if missing:
|
|
return (
|
|
"Yep. I can answer questions here without calling HN or DEV. "
|
|
"For the research report itself, I still need "
|
|
+ (missing[0] if len(missing) == 1 else ", ".join(missing[:-1]) + f", and {missing[-1]}")
|
|
+ "."
|
|
)
|
|
return (
|
|
"Yes. I have enough context for the report, and I'll only call HN and DEV.to when you ask me to find, recommend, or analyze trends and talk/blog ideas. "
|
|
"Until then, I can refine the angle, audience, competitors, or keywords conversationally."
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
def wants_memory_lookup(prompt: str) -> bool:
|
|
lowered = prompt.lower()
|
|
if wants_explicit_research_run(prompt):
|
|
return False
|
|
historical_research_phrases = (
|
|
"our research",
|
|
"research we have done",
|
|
"research we've done",
|
|
"research we did",
|
|
"we researched",
|
|
"we have researched",
|
|
"we've researched",
|
|
"done so far",
|
|
"so far",
|
|
"last few products",
|
|
"last few topics",
|
|
"last few ideas",
|
|
"last products",
|
|
"last topics",
|
|
"last ideas",
|
|
)
|
|
historical_subject_terms = (
|
|
"product",
|
|
"products",
|
|
"topic",
|
|
"topics",
|
|
"idea",
|
|
"ideas",
|
|
"article",
|
|
"articles",
|
|
"blog",
|
|
"blogs",
|
|
"talk",
|
|
"talks",
|
|
"publish",
|
|
"work on",
|
|
)
|
|
if any(phrase in lowered for phrase in historical_research_phrases) and any(
|
|
term in lowered for term in historical_subject_terms
|
|
):
|
|
return True
|
|
memory_terms = (
|
|
"before",
|
|
"past",
|
|
"previous",
|
|
"previously",
|
|
"recent",
|
|
"recently",
|
|
"lately",
|
|
"already",
|
|
"history",
|
|
"memory",
|
|
"discussed",
|
|
"researched",
|
|
"generated",
|
|
"suggested",
|
|
"recommend",
|
|
"recommended",
|
|
"found",
|
|
"so far",
|
|
)
|
|
subject_terms = (
|
|
"product",
|
|
"products",
|
|
"topic",
|
|
"topics",
|
|
"idea",
|
|
"ideas",
|
|
"article",
|
|
"articles",
|
|
"blog",
|
|
"blogs",
|
|
"talk",
|
|
"talks",
|
|
"brief",
|
|
"briefs",
|
|
"report",
|
|
"reports",
|
|
"about",
|
|
"result",
|
|
"results",
|
|
"suggestion",
|
|
"suggestions",
|
|
"recommendation",
|
|
"recommendations",
|
|
)
|
|
return any(term in lowered for term in memory_terms) and any(
|
|
term in lowered for term in subject_terms
|
|
)
|
|
|
|
|
|
def followup_response(prompt: str, result: AgentRunResult | None) -> str | None:
|
|
if result is None:
|
|
return None
|
|
|
|
lowered = prompt.lower()
|
|
summary_terms = (
|
|
"what topics",
|
|
"what ideas",
|
|
"what trends",
|
|
"what did we find",
|
|
"what we found",
|
|
"what did you find",
|
|
"what was suggested",
|
|
"what were suggested",
|
|
"topics were suggested",
|
|
"ideas were suggested",
|
|
"suggested recently",
|
|
"summary",
|
|
"recap",
|
|
)
|
|
if any(term in lowered for term in summary_terms):
|
|
lines = [result.report.summary.strip() or f"Here are the latest ideas for {result.report.company}."]
|
|
if result.report.trend_digest:
|
|
lines.append("")
|
|
lines.append("Top trends from the latest report:")
|
|
for index, trend in enumerate(result.report.trend_digest[:5], 1):
|
|
lines.append(f"{index}. **{trend.topic}** — intensity {trend.intensity_score}")
|
|
if result.report.content_ideas:
|
|
lines.append("")
|
|
lines.append("Top talk/blog ideas from the latest report:")
|
|
for index, idea in enumerate(result.report.content_ideas[:5], 1):
|
|
lines.append(f"{index}. **{idea.title}** ({idea.format}, score {idea.score})")
|
|
return "\n".join(lines)
|
|
|
|
if "evidence" in lowered and any(term in lowered for term in ("topic", "brief", "idea", "opportunity", "trend")):
|
|
topic_number = _first_number(lowered)
|
|
if topic_number is None:
|
|
return "Tell me which idea number you want evidence for."
|
|
index = topic_number - 1
|
|
if index < 0 or index >= len(result.report.content_ideas):
|
|
return f"I only have {len(result.report.content_ideas)} ideas in the latest report."
|
|
idea = result.report.content_ideas[index]
|
|
hn_links = "\n".join(f"- {url}" for url in idea.hn_evidence) or "- No strong HN evidence found in this run."
|
|
dev_links = "\n".join(f"- {url}" for url in idea.dev_links) or "- No specific DEV.to article matched this idea closely."
|
|
return (
|
|
f"### Evidence for idea {topic_number}: {idea.title}\n\n"
|
|
f"HN evidence:\n{hn_links}\n\n"
|
|
f"DEV supply links:\n{dev_links}\n\n"
|
|
f"DEV supply gap: {idea.dev_gap}"
|
|
)
|
|
|
|
if "more technical" in lowered or "technical" in lowered:
|
|
lines = ["### More technical angles"]
|
|
for index, idea in enumerate(result.report.content_ideas, 1):
|
|
outline = "; ".join(idea.outline[:3]) if idea.outline else "Add a runnable implementation demo."
|
|
lines.append(f"{index}. {idea.title}: {outline}")
|
|
return "\n".join(lines)
|
|
|
|
if "comparison" in lowered or "vs" in lowered:
|
|
lines = ["### Comparison reframing"]
|
|
for index, idea in enumerate(result.report.content_ideas, 1):
|
|
lines.append(
|
|
f"{index}. Reframe '{idea.title}' as evaluation criteria, tradeoffs, and migration risks."
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
return None
|
|
|
|
|
|
def _first_number(text: str) -> int | None:
|
|
for token in text.replace("#", " ").split():
|
|
if token.isdigit():
|
|
return int(token)
|
|
return None
|