ankn
e08c484838
Six-step weekly workflow (research → sources → storyline → draft → quality → publication) supporting Claude, ChatGPT, Gemini, and Mistral in parallel for creative steps. Web search via Anthropic tool for news research. Episode index built from 34 existing KNIEPUNKT episodes for redundancy checks. Sessions persisted as JSON for mid-workflow resume. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
"""Anthropic client and chat helpers."""
|
|
import os
|
|
from typing import Any
|
|
|
|
import anthropic
|
|
|
|
MODEL = "claude-sonnet-4-6"
|
|
MAX_TOKENS = 8096
|
|
|
|
|
|
def get_client() -> anthropic.Anthropic:
|
|
api_key = os.environ.get("ANTHROPIC_API_KEY")
|
|
if not api_key:
|
|
raise RuntimeError("ANTHROPIC_API_KEY environment variable not set.")
|
|
return anthropic.Anthropic(api_key=api_key)
|
|
|
|
|
|
def chat(
|
|
client: anthropic.Anthropic,
|
|
messages: list[dict],
|
|
system: str,
|
|
max_tokens: int = MAX_TOKENS,
|
|
) -> str:
|
|
response = client.messages.create(
|
|
model=MODEL,
|
|
max_tokens=max_tokens,
|
|
system=system,
|
|
messages=messages,
|
|
)
|
|
return _extract_text(response.content)
|
|
|
|
|
|
_MAX_SEARCH_ITERATIONS = 10
|
|
|
|
|
|
def chat_with_search(
|
|
client: anthropic.Anthropic,
|
|
messages: list[dict],
|
|
system: str,
|
|
) -> str:
|
|
"""Run a conversation with web search enabled. Loops on client-side tool_use only."""
|
|
current_messages = list(messages)
|
|
|
|
for _ in range(_MAX_SEARCH_ITERATIONS):
|
|
response = client.messages.create(
|
|
model=MODEL,
|
|
max_tokens=MAX_TOKENS,
|
|
system=system,
|
|
tools=[{"type": "web_search_20250305", "name": "web_search"}],
|
|
messages=current_messages,
|
|
)
|
|
|
|
if response.stop_reason == "end_turn":
|
|
return _extract_text(response.content)
|
|
|
|
if response.stop_reason == "tool_use":
|
|
current_messages.append({"role": "assistant", "content": response.content})
|
|
# Only handle client-side tool_use blocks; server_tool_use results are
|
|
# already embedded in the response and must not be echoed back.
|
|
tool_results = [
|
|
{"type": "tool_result", "tool_use_id": block.id, "content": ""}
|
|
for block in response.content
|
|
if getattr(block, "type", None) == "tool_use"
|
|
]
|
|
if tool_results:
|
|
current_messages.append({"role": "user", "content": tool_results})
|
|
else:
|
|
return _extract_text(response.content)
|
|
|
|
return _extract_text(response.content)
|
|
|
|
|
|
def _extract_text(content: list[Any]) -> str:
|
|
parts = []
|
|
for block in content:
|
|
if hasattr(block, "text"):
|
|
parts.append(block.text)
|
|
return "\n".join(parts)
|