import os
import json
import re
from dotenv import load_dotenv
from exa_py import Exa
from openai import OpenAI
load_dotenv()
exa = Exa(api_key=os.environ["EXA_API_KEY"])
exa.headers["x-exa-integration"] = "cerebras-integration"
cerebras = OpenAI(
api_key=os.environ["CEREBRAS_API_KEY"],
base_url="https://api.cerebras.ai/v1",
default_headers={"X-Cerebras-3rd-Party-Integration": "exa"},
)
sources = []
index_by_url = {}
def register(title, url):
if url not in index_by_url:
sources.append((title or url, url))
index_by_url[url] = len(sources)
return index_by_url[url]
def exa_search(query, type="auto", num_results=10, max_age_hours=None, **_):
contents = {"highlights": True}
if max_age_hours is not None:
contents["max_age_hours"] = max_age_hours
results = exa.search(query, type=type, num_results=num_results, contents=contents)
return "\n\n".join(
f"[{register(r.title, r.url)}] {r.title or r.url}\nURL: {r.url}\n{' '.join(r.highlights or [])}"
for r in results.results
)
available_tools = {"exa_search": exa_search}
tools = [
{
"type": "function",
"function": {
"name": "exa_search",
"description": "Search the web with Exa and get clean, ready-to-use results. Best for current information, news, facts, people, and companies. Returns numbered sources [n] with title, URL, and highlights.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "The search query."},
"type": {
"type": "string",
"enum": ["auto", "fast", "deep"],
"description": "Search strategy. 'auto' (default) balances quality and speed; 'fast' is the lowest-latency option; 'deep' is most thorough.",
},
"num_results": {
"type": "integer",
"description": "Number of results to return (1-100, default 10).",
},
"max_age_hours": {
"type": "integer",
"description": "Only accept cached pages newer than this many hours; older pages are refreshed before returning. Omit for no freshness limit, 0 to always fetch fresh content, or -1 to use cached content only.",
},
},
"required": ["query"],
},
},
}
]
# Remove stray citation markers (e.g. 【†L1-L9】) the model sometimes adds.
GARBAGE = re.compile(r"【[^】]*】|\d*†[^\s\]】]*】?|[【】†]")
def finalize(answer):
answer = GARBAGE.sub("", answer)
answer = re.sub(r"\[\[(\d+)\]\]", r"[\1]", answer).strip()
if not sources:
return answer
lines = "\n".join(f"[{i}] {title} - {url}" for i, (title, url) in enumerate(sources, 1))
return f"{answer}\n\nSources:\n{lines}"
def answer_with_search(question):
messages = [
{
"role": "system",
"content": (
"You are a research assistant. Use exa_search to find current information, then "
"answer the question. Cite sources inline as [n], matching the labels returned "
"by exa_search (for example [1] or [2])."
),
},
{"role": "user", "content": question},
]
for _ in range(6):
response = cerebras.chat.completions.create(
model="gpt-oss-120b",
messages=messages,
tools=tools,
tool_choice="auto",
max_completion_tokens=2000,
)
message = response.choices[0].message
messages.append(message)
if not message.tool_calls:
return finalize(message.content or "")
for tool_call in message.tool_calls:
tool_fn = available_tools.get(tool_call.function.name)
try:
args = json.loads(tool_call.function.arguments)
result = tool_fn(**args) if tool_fn else f"Unknown tool: {tool_call.function.name}"
except Exception as e:
result = f"Tool error ({type(e).__name__}): {e}. Adjust your arguments and try again."
messages.append({"role": "tool", "tool_call_id": tool_call.id, "content": result})
return "Could not produce a final answer within the step limit."
print(answer_with_search("What are the latest AI model releases, and what makes them notable?"))
