Spaces:
Runtime error
Runtime error
| import json | |
| import os | |
| import re | |
| from json import JSONDecoder | |
| from typing import List, Dict | |
| from dotenv import load_dotenv | |
| from pydantic import ValidationError | |
| from langchain_core.messages import AIMessage, HumanMessage | |
| from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder | |
| from langchain_openai import ChatOpenAI | |
| from models import AgentLLMOutput, ChatResponse, Memory | |
| from memory_store import MemoryStore | |
| load_dotenv() | |
| def _openrouter_referer() -> str: | |
| if host := os.getenv("SPACE_HOST"): | |
| return host | |
| if space_id := os.getenv("SPACE_ID"): | |
| return f"https://huggingface.co/spaces/{space_id}" | |
| return "http://localhost:8000" | |
| SYSTEM_PROMPT = """\ | |
| You are a customer service assistant for an internet, TV, and telephony provider in Brazil. | |
| This is a memory-observability PoC: you help the current customer resolve their situation using \ | |
| what past customers said in previous conversations — stored as memories below. | |
| Memories are the customers' own words (or faithful paraphrase), NOT official policy, CRM data, \ | |
| or verified operational facts. Use them to empathize, recognize patterns, and suggest helpful \ | |
| approaches that worked before — with caution when memories conflict or may be outdated. | |
| {memory_context} | |
| ───────────────────────────────────────────── | |
| RESPONSE FORMAT — you MUST return valid JSON only, no other text: | |
| {{ | |
| "response": "Your reply to the customer in Brazilian Portuguese (clear, empathetic, actionable)", | |
| "memories_used": ["id1", "id2"], | |
| "new_memories": [ | |
| {{ | |
| "content": "What the current customer said, preserved in their voice as closely as possible", | |
| "type": "episodic|semantic|state|procedural (English only, not episodico/semantico)", | |
| "context_tags": ["tag1", "tag2"], | |
| "summary": "5-10 word summary for display" | |
| }} | |
| ] | |
| }} | |
| ───────────────────────────────────────────── | |
| MEMORY TYPES (content is usually a past customer's statement): | |
| - episodic → specific situation a customer reported ("my portability has been stuck for 5 days") | |
| - semantic → recurring pattern from multiple customers ("new installs often question real speed") | |
| - state → recent claim about current conditions ("the app won't load my boleto since yesterday") | |
| - procedural → lesson from how support went ("asking cable vs Wi-Fi before sending a tech helped") | |
| ───────────────────────────────────────────── | |
| GUIDELINES: | |
| - Always respond to the customer in Brazilian Portuguese; be empathetic and avoid unexplained jargon | |
| - Use only memories genuinely relevant to the current message | |
| - List only the IDs of memories you actually drew on in your response | |
| - When memories contradict each other, do NOT state uncertain things as fact; ask clarifying questions \ | |
| or acknowledge uncertainty | |
| - Create new_memories only for noteworthy things the CURRENT customer said — keep their wording \ | |
| and tone; not every message needs a new memory | |
| - Choose memory type based on what was said: one-off event (episodic), recurring theme (semantic), \ | |
| current-sounding situation (state), or insight about what helped/hurt in support (procedural) | |
| - PoC limits: you have no access to billing, CRM, or network systems — guide with questions, \ | |
| logical troubleshooting steps, and reasonable next steps without inventing protocol numbers, \ | |
| discounts, stock levels, or coverage | |
| - Respond ONLY with the JSON object — no preamble, no markdown fences | |
| - Previous assistant turns in the chat history are plain-text summaries for context; \ | |
| your current reply must still be ONLY the JSON object, never duplicate the answer outside JSON | |
| """ | |
| OPENROUTER_API_URL = os.getenv("OPENROUTER_API_URL") | |
| MODEL = os.getenv("OPENROUTER_MODEL") | |
| PROMPT = ChatPromptTemplate.from_messages([ | |
| ("system", SYSTEM_PROMPT), | |
| MessagesPlaceholder("history"), | |
| ("human", "{input}"), | |
| ]) | |
| def _parse_agent_llm_output(text: str) -> AgentLLMOutput: | |
| """Accept strict JSON or model output with prose before/after the JSON object.""" | |
| text = (text or "").strip() | |
| if not text: | |
| raise ValueError("Empty LLM output") | |
| decoder = JSONDecoder() | |
| candidates: List[str] = [text] | |
| for match in re.finditer( | |
| r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL | re.IGNORECASE | |
| ): | |
| candidates.append(match.group(1)) | |
| for raw in candidates: | |
| try: | |
| return AgentLLMOutput.model_validate(json.loads(raw)) | |
| except (json.JSONDecodeError, ValueError, ValidationError): | |
| continue | |
| start = 0 | |
| while True: | |
| brace = text.find("{", start) | |
| if brace == -1: | |
| break | |
| try: | |
| obj, _ = decoder.raw_decode(text, brace) | |
| if isinstance(obj, dict) and "response" in obj: | |
| return AgentLLMOutput.model_validate(obj) | |
| except (json.JSONDecodeError, ValidationError): | |
| pass | |
| start = brace + 1 | |
| raise ValueError("No valid AgentLLMOutput JSON found in model response") | |
| class Agent: | |
| def __init__(self, memory_store: MemoryStore): | |
| self.memory_store = memory_store | |
| llm = ChatOpenAI( | |
| base_url=OPENROUTER_API_URL, | |
| api_key=os.getenv("OPENROUTER_API_KEY"), | |
| model=MODEL, | |
| max_tokens=1500, | |
| timeout=60.0, | |
| extra_body={ | |
| "chat_template_kwargs": {"enable_thinking": False}, | |
| "response_format": {"type": "json_object"}, | |
| }, | |
| default_headers={ | |
| "HTTP-Referer": _openrouter_referer(), | |
| "X-Title": "Agent Memory Phase 1", | |
| }, | |
| ) | |
| self.chain = PROMPT | llm | |
| async def chat( | |
| self, | |
| message: str, | |
| conversation_history: List[Dict[str, str]], | |
| ) -> ChatResponse: | |
| relevant = self.memory_store.search(message, n_results=6) | |
| state_mems = self.memory_store.search(message, n_results=3, type_filter="state") | |
| seen: set = set() | |
| candidates: List[Memory] = [] | |
| for m in relevant + state_mems: | |
| if m.id not in seen: | |
| seen.add(m.id) | |
| candidates.append(m) | |
| memory_context = self._format_memories(candidates) | |
| history = [ | |
| HumanMessage(content=turn["content"]) | |
| if turn["role"] == "user" | |
| else AIMessage(content=turn["content"]) | |
| for turn in conversation_history[-6:] | |
| ] | |
| raw = await self.chain.ainvoke({ | |
| "memory_context": memory_context, | |
| "input": message, | |
| "history": history, | |
| }) | |
| parsed = _parse_agent_llm_output(raw.content) | |
| for mem_id in parsed.memories_used: | |
| self.memory_store.update_access(mem_id) | |
| new_memories_saved: List[Memory] = [] | |
| for nm in parsed.new_memories: | |
| saved = self.memory_store.add_memory( | |
| content=nm.content, | |
| memory_type=nm.type.value, | |
| source="agent", | |
| context_tags=nm.context_tags, | |
| summary=nm.summary, | |
| ) | |
| new_memories_saved.append(saved) | |
| return ChatResponse( | |
| response=parsed.response, | |
| memories_used=parsed.memories_used, | |
| new_memories=new_memories_saved, | |
| all_memories=self.memory_store.list_memories(), | |
| ) | |
| def _format_memories(self, memories: List[Memory]) -> str: | |
| if not memories: | |
| return "(no memories available)" | |
| lines = [] | |
| for m in memories: | |
| tags = ", ".join(m.context_tags) if m.context_tags else "—" | |
| lines.append( | |
| f"[ID: {m.id}] [{m.type.upper()}] {m.content}\n" | |
| f" tags: {tags} | score: {m.relevance_score:.2f} | accessed: {m.access_count}x" | |
| ) | |
| return "\n\n".join(lines) | |