# Support resolution agent
> [!NOTE]
> Code available [here](https://github.com/unionai/unionai-examples/tree/main/v2/tutorials/support_resolution_agent).
This example demonstrates how to build a customer-support and field-service resolution agent on Flyte. The agent resolves tickets that need current public information — return policies, weather advisories, product recalls, manufacturer specs — and drafts a customer-ready reply with sources a human agent can verify before sending.
The [You.com Research API](https://you.com/docs/research/overview) grounds each ticket in fresh, citable sources. [Claude](https://docs.anthropic.com/) via [LiteLLM](https://docs.litellm.ai/) turns that research into a reply draft. With `research_effort="lite"`, the research step stays fast enough for human-in-the-loop support flows.
Flyte provides:
- **Fan-out parallelism** across support tickets
- **`@flyte.trace`** on every external call for lineage
- A **two-step pipeline** per ticket: ground the answer, then draft the reply
- **Flyte reports** with draft replies and verifiable source citations
## Setting up the environment
The agent runs in a `TaskEnvironment` with secrets for the You.com and Anthropic API keys and a container image built from the `uv` script dependencies.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f"
"
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
The Python packages are declared at the top of the file using the `uv` script style:
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# ///
```
## Data types
Each `Ticket` carries a ticket ID, a customer question, and optional product or vendor context. The final `Resolution` includes the grounded answer, a draft reply, and the list of You.com sources.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f""
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
## Ground answers with the You.com Research API
The `you_research` helper calls the [You.com Research API](https://you.com/docs/research/overview) with a configurable `research_effort`. For support use cases, `lite` provides a fast, citation-backed answer suitable for real-time, human-in-the-loop flows. See the [Research API reference](https://you.com/docs/api-reference/research/v1-research) for effort levels and parameters.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f""
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
## Ground one ticket
The `ground_answer` task combines the ticket question and context into a research query and collects the grounded answer plus structured sources from the Research API response.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f""
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
## Draft a customer-ready reply
The `draft_reply` task turns the grounded answer into a concise, friendly reply that cites source URLs inline so a human agent can verify before sending.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f""
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
## Resolve one ticket
Each ticket runs `ground_answer` followed by `draft_reply` in sequence.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f""
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
## Orchestration
The `support_resolution` driver task fans out across all tickets and renders a Flyte report with every draft reply and its sources.
```
# /// script
# requires-python = "==3.13"
# dependencies = [
# "flyte>=2.4.0",
# "httpx>=0.27.0",
# "litellm>=1.72.0",
# ]
# main = "support_resolution"
# params = ""
# ///
"""Customer-support & field-service resolution agent.
Grounds a support ticket in fresh, public, citable sources via the You.com
Research API (low effort for low latency, human-in-the-loop use), then uses
Claude to draft a customer-ready reply that cites its sources inline so a human
agent can verify before sending.
"""
# {{docs-fragment env}}
import asyncio
import json
import os
from dataclasses import dataclass, field
import flyte
MODEL = "anthropic/claude-haiku-4-5"
env = flyte.TaskEnvironment(
name="support-resolution",
secrets=[
flyte.Secret(key="youdotcom-api-key", as_env_var="YOU_API_KEY"),
flyte.Secret(key="internal-anthropic-api-key", as_env_var="ANTHROPIC_API_KEY"),
],
image=flyte.Image.from_uv_script(__file__, name="support-resolution", pre=True),
resources=flyte.Resources(cpu="1", memory="1Gi"),
)
# {{/docs-fragment env}}
# {{docs-fragment data_types}}
@dataclass
class Source:
title: str
url: str
snippet: str
domain: str = ""
favicon: str = ""
def _domain(url: str) -> str:
from urllib.parse import urlparse
try:
return urlparse(url).netloc.replace("www.", "")
except Exception:
return ""
def _favicon_for(url: str) -> str:
return f"https://ydc-index.io/favicon?domain={_domain(url)}&size=128"
@dataclass
class Ticket:
ticket_id: str
question: str
context: str = ""
@dataclass
class Grounding:
answer: str
sources: list[Source] = field(default_factory=list)
@dataclass
class Resolution:
ticket_id: str
ticket: str
grounded_answer: str
draft_reply: str
sources: list[Source] = field(default_factory=list)
@dataclass
class ResolutionReport:
resolutions: list[Resolution] = field(default_factory=list)
# {{/docs-fragment data_types}}
# {{docs-fragment you_research}}
YOU_RESEARCH_URL = "https://api.you.com/v1/research"
async def _you_post(url: str, body: dict, timeout: float = 120.0) -> dict:
"""POST with exponential backoff + jitter on 429 rate limits."""
import random
import httpx
headers = {
"X-API-Key": os.environ["YOU_API_KEY"],
"Content-Type": "application/json",
}
async with httpx.AsyncClient(timeout=timeout) as client:
for attempt in range(7):
resp = await client.post(url, headers=headers, json=body)
if resp.status_code == 429 and attempt < 6:
wait = float(resp.headers.get("retry-after") or 0) or min(2**attempt, 30)
await asyncio.sleep(wait + random.uniform(0, 2))
continue
resp.raise_for_status()
return resp.json()
resp.raise_for_status()
return resp.json()
@flyte.trace
async def you_research(question: str, research_effort: str = "lite") -> dict:
"""Fast, citation-backed grounding for a support question."""
body = {"input": question, "research_effort": research_effort}
return await _you_post(YOU_RESEARCH_URL, body)
# {{/docs-fragment you_research}}
# {{docs-fragment ground_answer}}
@env.task(retries=3)
async def ground_answer(ticket: str, context: str, research_effort: str) -> Grounding:
"""Ground the ticket in fresh public sources via the Research API."""
question = ticket if not context else f"{ticket}\n\nContext: {context}"
result = await you_research(question, research_effort)
output = result.get("output", {})
answer = output.get("content", "")
if not isinstance(answer, str):
answer = json.dumps(answer)
sources = []
for s in output.get("sources", []) or []:
url = str(s.get("url", ""))
sources.append(
Source(
title=str(s.get("title", "") or url),
url=url,
snippet=str((s.get("snippets") or [""])[0]),
domain=_domain(url),
favicon=_favicon_for(url),
)
)
return Grounding(answer=answer, sources=sources)
# {{/docs-fragment ground_answer}}
# {{docs-fragment draft_reply}}
@flyte.trace
async def _draft(ticket: str, answer: str, sources_text: str) -> str:
from litellm import acompletion
system = (
"You are a senior customer-support agent. Using ONLY the grounded "
"answer and sources provided, draft a concise, friendly, customer-ready "
"reply. Cite the relevant source URL inline in parentheses after any "
"factual claim so a human agent can verify before sending. If the "
"sources do not answer the question, say so plainly."
)
user = (
f"Customer ticket: {ticket}\n\n"
f"Grounded answer:\n{answer}\n\nSources:\n{sources_text}"
)
resp = await acompletion(
model=MODEL,
messages=[
{"role": "system", "content": system},
{"role": "user", "content": user},
],
temperature=0.2,
max_tokens=1024,
)
return resp.choices[0].message.content
@env.task
async def draft_reply(ticket: Ticket, grounding: Grounding) -> Resolution:
"""Turn the grounded answer into a cited, customer-ready reply."""
sources_text = "\n".join(
f"- {s.title} ({s.domain}): {s.url}\n \"{s.snippet}\""
for s in grounding.sources
)
reply = await _draft(ticket.question, grounding.answer, sources_text)
return Resolution(
ticket_id=ticket.ticket_id,
ticket=ticket.question,
grounded_answer=grounding.answer,
draft_reply=reply,
sources=grounding.sources,
)
# {{/docs-fragment draft_reply}}
# {{docs-fragment resolve_ticket}}
async def resolve_ticket(ticket: Ticket, research_effort: str) -> Resolution:
"""Ground one ticket then draft its reply."""
grounding = await ground_answer(ticket.question, ticket.context, research_effort)
return await draft_reply(ticket, grounding)
# {{/docs-fragment resolve_ticket}}
# {{docs-fragment report}}
REPORT_CSS = """
"""
def _cite(s: Source) -> str:
"""Render a rich You.com Research citation for a support source."""
if not s.url:
return ""
snip = f"“{s.snippet}”
" if s.snippet else ""
return (
f""
f"
")
cards.append(
f""
f"
{res.ticket_id}
"
f"
{res.ticket}
"
f"
Draft reply (for human review)
{reply_html}"
+ (f"
You.com sources ({len(res.sources)})
{src}" if src else "")
+ "
Support Resolutions
Tickets grounded in fresh public sources via the You.com
Research API — draft replies cite sources a human agent can verify.
{len(report.resolutions)} tickets
{total_sources} You.com sources cited
{''.join(cards) or "
No tickets processed.
"}
Each ticket grounded by the You.com Research API
(lite effort for low-latency, human-in-the-loop use). Sources
include domain, title, and snippet provenance — ready to paste into a
customer reply with verification links.
"""
# {{/docs-fragment report}}
# {{docs-fragment driver}}
def _default_tickets() -> list[Ticket]:
return [
Ticket(
"tkt-1",
"Is there a recall on the DeWalt DCD777 cordless drill, and what should "
"the customer do if there is?",
"Customer purchased the drill recently and is asking about safety recalls.",
),
Ticket(
"tkt-2",
"What is Sony's current return policy for the WH-1000XM5 headphones?",
"Customer wants to return an opened pair bought 20 days ago.",
),
Ticket(
"tkt-3",
"Are there any current weather advisories that could delay flights out of "
"Denver International Airport today?",
"Customer is worried about a connecting flight.",
),
Ticket(
"tkt-4",
"What are the dimensions and weight capacity of the IKEA BEKANT desk?",
"Customer is checking if it fits their space before resolving a complaint.",
),
Ticket(
"tkt-5",
"Has Samsung issued any recall or safety notice for the Galaxy Z Fold5?",
"Customer reports overheating and wants to know about known issues.",
),
Ticket(
"tkt-6",
"What is the warranty period for a Dyson V15 Detect vacuum in the US?",
"Customer's vacuum stopped working and asks about coverage.",
),
]
@env.task(report=True)
async def support_resolution(
tickets: list[Ticket] | None = None,
research_effort: str = "lite",
) -> ResolutionReport:
"""Fan out across support tickets, grounding and drafting cited replies."""
if tickets is None:
tickets = _default_tickets()
with flyte.group("resolve-tickets"):
resolutions = await asyncio.gather(
*[resolve_ticket(t, research_effort) for t in tickets]
)
report = ResolutionReport(resolutions=list(resolutions))
await flyte.report.replace.aio(_render_report(report), do_flush=True)
await flyte.report.flush.aio()
return report
# {{/docs-fragment driver}}
# {{docs-fragment main}}
if __name__ == "__main__":
flyte.init_from_config()
run = flyte.run(support_resolution)
print(run.url)
run.wait()
# {{/docs-fragment main}}
```
*Source: https://github.com/unionai/unionai-examples/blob/main/v2/tutorials/support_resolution_agent/main.py*
## Run the agent
### Create secrets
Get a You.com API key from the [You.com platform](https://you.com/platform) (see the [quickstart guide](https://you.com/docs/quickstart)). Get an Anthropic API key from the [Anthropic console](https://console.anthropic.com/).
Register both keys as Flyte secrets. The secret key names must match those declared in the `TaskEnvironment`:
```
flyte create secret youdotcom-api-key
flyte create secret internal-anthropic-api-key
```
See [Secrets](https://www.union.ai/docs/v2/union/user-guide/task-configuration/secrets/page.md) for scoping and file-based secrets.
### Run locally or remotely
From the [example directory](https://github.com/unionai/unionai-examples/tree/main/v2/tutorials/support_resolution_agent):
```
cd v2/tutorials/support_resolution_agent
uv run --script main.py
```
To test locally without Flyte secrets:
```
export YOU_API_KEY=
export ANTHROPIC_API_KEY=
uv run --script main.py
```
When the run completes, open the Flyte report to review draft replies for each ticket, with You.com source citations ready for a human agent to verify and paste into a customer response.
---
**Source**: https://github.com/unionai/unionai-docs/blob/main/content/tutorials/agents/support-resolution-agent/_index.md
**HTML**: https://www.union.ai/docs/v2/union/tutorials/agents/support-resolution-agent/