File size: 4,420 Bytes
b6ed6f7 0aa8e87 b6ed6f7 0aa8e87 b6ed6f7 18ddfa2 b6ed6f7 0aa8e87 18ddfa2 0aa8e87 18ddfa2 0aa8e87 b6ed6f7 0aa8e87 18ddfa2 0aa8e87 18ddfa2 0aa8e87 18ddfa2 0aa8e87 18ddfa2 0aa8e87 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from src.agents.CallState import CallState
from src.agents.schemas import CallSummary
class SummarizationAgent:
def __init__(self):
self.llm = ChatOpenAI(model="gpt-4o", temperature=0)
def __call__(self, state: CallState) -> CallState:
"""Generates summaries and key points."""
clean_text = state.get("clean_content", state.get("content", ""))
if not clean_text:
return state
prompt = PromptTemplate.from_template(
"Summarize the following call transcript and extract key points.\n"
"Return a concise summary, a short list of key points, 2-8 action items, 3-8 topic tags, and 2-6 highlights.\n"
"Action items must be concrete follow-ups; include the owner (Agent/Customer) when you can.\n\n"
"Transcript:\n{text}"
)
structured_llm = self._structured_llm()
chain = prompt | structured_llm
result = chain.invoke({"text": clean_text})
if isinstance(result, CallSummary):
state["summary"] = result.summary
state["key_points"] = result.key_points
state["action_items"] = result.action_items
state["tags"] = result.tags
state["highlights"] = result.highlights
else:
state["summary"] = (result or {}).get("summary", "")
state["key_points"] = (result or {}).get("key_points", [])
state["action_items"] = (result or {}).get("action_items", [])
state["tags"] = (result or {}).get("tags", [])
state["highlights"] = (result or {}).get("highlights", [])
return state
def _structured_llm(self):
# Some LangChain versions support different output enforcement methods.
for kwargs in ({"method": "function_calling"}, {"method": "json_mode"}, {}):
try:
return self.llm.with_structured_output(CallSummary, **kwargs)
except TypeError:
continue
return self.llm.with_structured_output(CallSummary)
@staticmethod
def fallback(state: CallState) -> CallState:
text = state.get("clean_content") or state.get("content") or ""
if not text:
return state
# Best-effort retry using structured output with alternate enforcement methods.
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = PromptTemplate.from_template(
"Summarize the following call transcript.\n"
"Return: summary, key_points, action_items, tags, highlights.\n\n"
"Transcript:\n{text}"
)
for kwargs in ({"method": "function_calling"}, {"method": "json_mode"}, {}):
try:
structured_llm = llm.with_structured_output(CallSummary, **kwargs)
result = (prompt | structured_llm).invoke({"text": text})
if isinstance(result, CallSummary):
state["summary"] = result.summary
state["key_points"] = result.key_points
state["action_items"] = result.action_items
state["tags"] = result.tags
state["highlights"] = result.highlights
return state
state["summary"] = (result or {}).get("summary", state.get("summary", ""))
state["key_points"] = (result or {}).get("key_points", state.get("key_points") or [])
state["action_items"] = (result or {}).get("action_items", state.get("action_items") or [])
state["tags"] = (result or {}).get("tags", state.get("tags") or [])
state["highlights"] = (result or {}).get("highlights", state.get("highlights") or [])
return state
except Exception:
continue
# Last resort heuristic fallback that avoids breaking the UI.
summary = (text.strip()[:800] + ("…" if len(text.strip()) > 800 else "")).strip()
state["summary"] = summary or state.get("summary", "")
state["key_points"] = state.get("key_points") or []
state["action_items"] = state.get("action_items") or []
state["tags"] = state.get("tags") or []
state["highlights"] = state.get("highlights") or []
return state
|