"""Voice research agent: Cartesia Line + Tavily."""
from datetime import datetime
import os
from typing import Annotated, Optional
from tavily import AsyncTavilyClient
from line.llm_agent import LlmAgent, LlmConfig, ToolEnv, end_call, loopback_tool
from line.voice_agent_app import AgentEnv, CallRequest, VoiceAgentApp
SYSTEM_PROMPT = """Today is {today}. You are a research assistant on a live voice call.
You have two web tools:
1. web_search — Find relevant pages. Use for current events, prices, news, or anything
that needs fresh data. Start here.
2. web_extract — Pull full content from a specific URL. Use only when a search snippet
is too thin, or the user names a link they want read.
Lead with the answer. Two or three sentences unless the user asks for more. Name your
source when it matters. If results conflict or look stale, say so. Use end_call when
the user wraps up.
This is a voice call. Plain sentences only — no markdown, no lists, no special characters."""
INTRODUCTION = (
"Hey! I'm your research assistant, powered by Tavily and Cartesia. "
"Ask me anything and I'll dig it up live."
)
EXTRACT_MAX_CHARS = 3000
class TavilyTools:
# One client per agent so the httpx connection pool is reused across tool calls.
def __init__(self, api_key: str):
self._client = AsyncTavilyClient(api_key=api_key, client_source="cartesia-line-agent")
@loopback_tool
async def web_search(
self,
ctx: ToolEnv,
query: Annotated[str, "The search query. Be specific."],
time_range: Annotated[
Optional[str],
"Optional recency filter: 'day', 'week', 'month', or 'year'.",
] = None,
) -> str:
"""Search the web for current information."""
# 'fast' depth keeps latency in the budget for a voice turn.
kwargs: dict = {"query": query, "search_depth": "fast", "max_results": 5}
if time_range is not None:
kwargs["time_range"] = time_range
response = await self._client.search(**kwargs)
results = response.get("results", [])
if not results:
return "No relevant information found."
parts = [f"Search results for: '{query}'\n"]
for i, r in enumerate(results, start=1):
parts.append(f"\n--- Source {i}: {r['title']} (score {r.get('score', 0):.2f}) ---\n")
if r.get("content"):
parts.append(f"{r['content']}\n")
parts.append(f"URL: {r['url']}\n")
return "".join(parts)
@loopback_tool
async def web_extract(
self,
ctx: ToolEnv,
url: Annotated[str, "The URL to extract content from."],
) -> str:
"""Extract the full content of a webpage given its URL."""
response = await self._client.extract(urls=[url])
results = response.get("results", [])
if not results:
failed = response.get("failed_results", [])
if failed:
return f"Extraction failed for {url}: {failed[0].get('error', 'unknown error')}"
return "No content could be extracted from that URL."
raw = results[0].get("raw_content", "")
if not raw:
return "The page was reached but no readable content was found."
if len(raw) > EXTRACT_MAX_CHARS:
raw = raw[:EXTRACT_MAX_CHARS] + "\n\n[Content truncated]"
return f"Extracted content from {url}:\n\n{raw}"
def _require_env(name: str, where: str) -> str:
value = os.environ.get(name)
if not value:
raise RuntimeError(f"{name} is not set. Get one at {where} and export it before running.")
return value
async def get_agent(env: AgentEnv, call_request: CallRequest):
tavily = TavilyTools(api_key=_require_env("TAVILY_API_KEY", "https://app.tavily.com/home"))
return LlmAgent(
model="openai/gpt-4o-mini",
api_key=_require_env("OPENAI_API_KEY", "https://platform.openai.com/api-keys"),
tools=[tavily.web_search, tavily.web_extract, end_call],
config=LlmConfig(
system_prompt=SYSTEM_PROMPT.format(today=datetime.now().strftime("%Y-%m-%d")),
introduction=INTRODUCTION,
max_tokens=600,
temperature=0.7,
),
)
app = VoiceAgentApp(get_agent=get_agent)
if __name__ == "__main__":
app.run()
