feat: Add MCP integration, DeepSeek, Groq, and Playwright tests; update RAG agents and prompts

Author: gitwalter

agents/agile_factory/state/agile_state.py

@@ -9,7 +9,7 @@
 - Control fields (status, errors, thread_id, current_node)
 """
 
-try:d
+try:
     from typing_extensions import TypedDict
 except ImportError:
     from typing import TypedDict  # Python 3.12+

agents/rag/query_analyst_agent.py

@@ -217,14 +217,30 @@ async def _analyze_query_with_llm(self, query: str) -> Dict[str, Any]:
             logger.warning("LLM response not valid JSON, using fallback")
             return self._fallback_analysis(query)
 
+        # Safely parse complexity
+        complexity_val = analysis_data.get('complexity', 0.5)
+        try:
+            if isinstance(complexity_val, (int, float)):
+                complexity = float(complexity_val)
+            elif isinstance(complexity_val, str):
+                if complexity_val.lower() == 'low': complexity = 0.3
+                elif complexity_val.lower() == 'medium': complexity = 0.5
+                elif complexity_val.lower() == 'high': complexity = 0.8
+                else:
+                    complexity = float(complexity_val)
+            else:
+                complexity = 0.5
+        except (ValueError, TypeError):
+            complexity = 0.5
+
         # Ensure all fields are present
         analysis = {
             'original_query': query,
             'intent': analysis_data.get('intent', 'factual'),
             'rewritten_queries': analysis_data.get('rewritten_queries', [query]),
             'key_concepts': analysis_data.get('key_concepts', []),
             'search_strategy': analysis_data.get('search_strategy', 'focused'),
-            'complexity': float(analysis_data.get('complexity', 0.5)),
+            'complexity': complexity,
             'reasoning': analysis_data.get('reasoning', '')
         }
 

agents/rag/rag_swarm_coordinator.py

@@ -675,7 +675,7 @@ def _extract_query_from_state(self, state: MessagesState) -> Optional[str]:
             logger.warning(f"   First message: {str(messages[0])[:100]}")
         return None
 
-    def _query_analyst_node(self, state: MessagesState):
+    async def _query_analyst_node(self, state: MessagesState):
         """
         Query Analysis Node - Sophisticated query understanding.
         
@@ -719,9 +719,8 @@ def _query_analyst_node(self, state: MessagesState):
         logger.info(f"🔍 QueryAnalystAgent analyzing: '{query[:60]}...'")
 
         # Call QueryAnalystAgent
-        import asyncio
         task = {"query": query}
-        analysis_result = asyncio.run(self.query_analyst.execute(task))
+        analysis_result = await self.query_analyst.execute(task)
 
         # Format analysis as system message
         analysis_text = analysis_result.get("output_data", {})
@@ -736,7 +735,7 @@ def _query_analyst_node(self, state: MessagesState):
         logger.info(f"✅ Query analysis complete: {analysis_text.get('intent', 'unknown')} intent")
         return {"messages": [analysis_msg]}
 
-    def _retrieval_specialist_node(self, state: MessagesState):
+    async def _retrieval_specialist_node(self, state: MessagesState):
         """
         Retrieval Specialist Node - Multi-source retrieval orchestration.
         
@@ -756,9 +755,8 @@ def _retrieval_specialist_node(self, state: MessagesState):
 
         # Call QueryAnalystAgent first to get proper query_analysis format
         # (RetrievalSpecialistAgent expects query_analysis dict)
-        import asyncio
         query_task = {"query": query}
-        query_analysis_result = asyncio.run(self.query_analyst.execute(query_task))
+        query_analysis_result = await self.query_analyst.execute(query_task)
 
         # Extract query_analysis from QueryAnalystAgent result
         # QueryAnalystAgent returns: {'status': 'success', 'analysis': {...}, ...}
@@ -800,7 +798,7 @@ def _retrieval_specialist_node(self, state: MessagesState):
             task["document_filters"] = state["document_filters"]
             logger.info(f"🎯 Applying document filters from state: {state['document_filters']}")
 
-        retrieval_result = asyncio.run(self.retrieval_specialist.execute(task))
+        retrieval_result = await self.retrieval_specialist.execute(task)
 
         # Extract documents from search_results (correct field name)
         retrieved_docs = retrieval_result.get("search_results", [])
@@ -829,7 +827,7 @@ def _retrieval_specialist_node(self, state: MessagesState):
         logger.info(f"✅ Retrieved {len(tool_msgs)} documents from {len(set(d.name for d in tool_msgs))} sources")
         return {"messages": tool_msgs}
 
-    def _document_grader_node(self, state: MessagesState):
+    async def _document_grader_node(self, state: MessagesState):
         """
         Document Grader Node - Grade documents for relevance BEFORE re-ranking.
         
@@ -848,7 +846,7 @@ def _document_grader_node(self, state: MessagesState):
 
         # Get retrieval results (tool messages)
         tool_results = []
-        for msg in messages:
+        for msg in state["messages"]:  # Use state["messages"] as messages var wasn't defined in original code snippet but extracted above
             if isinstance(msg, ToolMessage):
                 tool_results.append({
                     'content': msg.content,
@@ -881,7 +879,7 @@ def _document_grader_node(self, state: MessagesState):
 Give a binary score 'yes' or 'no' to indicate whether the document is relevant to the question.
 Be lenient - if the document contains ANY relevant information, mark it as 'yes'."""
 
-                grade = llm.with_structured_output(GradeDocuments).invoke([
+                grade = await llm.with_structured_output(GradeDocuments).ainvoke([
                     HumanMessage(content=grade_prompt)
                 ])
 
@@ -917,7 +915,7 @@ def _document_grader_node(self, state: MessagesState):
         logger.info(f"✅ Grading complete: {len(graded_tool_msgs)}/{len(tool_results)} documents relevant")
         return {"messages": [grading_summary] + graded_tool_msgs}
 
-    def _re_ranker_node(self, state: MessagesState):
+    async def _re_ranker_node(self, state: MessagesState):
         """
         Re-Ranker Node - Relevance scoring and filtering.
         
@@ -948,12 +946,11 @@ def _re_ranker_node(self, state: MessagesState):
         logger.info(f"🎯 ReRankerAgent re-ranking {len(tool_results)} documents...")
 
         # Call ReRankerAgent
-        import asyncio
         task = {
             "query": query,
             "documents": tool_results
         }
-        rerank_result = asyncio.run(self.re_ranker.execute(task))
+        rerank_result = await self.re_ranker.execute(task)
 
         # Extract re-ranking info
         reranked_data = rerank_result.get("output_data", {})
@@ -972,7 +969,7 @@ def _re_ranker_node(self, state: MessagesState):
         logger.info(f"✅ Re-ranking complete: avg={avg_score:.1%}, quality_ok={quality_ok}")
         return {"messages": [rerank_msg]}
 
-    def _context_enrichment_node(self, state: MessagesState):
+    async def _context_enrichment_node(self, state: MessagesState):
         """
         Context Enrichment Node - Enrich context after re-ranking.
         
@@ -992,7 +989,7 @@ def _context_enrichment_node(self, state: MessagesState):
 
         # Get re-ranked context (tool messages from grader/re-ranker)
         context_parts = []
-        for msg in messages:
+        for msg in state["messages"]:  # Use state["messages"]
             if isinstance(msg, ToolMessage):
                 context_parts.append(msg.content)
 
@@ -1028,7 +1025,7 @@ def _context_enrichment_node(self, state: MessagesState):
 - Recommendation: "sufficient" if context is complete, "needs_more" if critical gaps exist"""
 
         try:
-            enrichment_response = llm.invoke([HumanMessage(content=enrichment_prompt)])
+            enrichment_response = await llm.ainvoke([HumanMessage(content=enrichment_prompt)])
             enriched_analysis = enrichment_response.content
 
             # Check if enrichment recommends more retrieval
@@ -1054,7 +1051,7 @@ def _context_enrichment_node(self, state: MessagesState):
             logger.error(f"❌ Context enrichment failed: {e}")
             return {"messages": [SystemMessage(content=f"Context enrichment error: {e}")]}
 
-    def _quality_assurance_node(self, state: MessagesState):
+    async def _quality_assurance_node(self, state: MessagesState):
         """
         Quality Assurance Node - Comprehensive quality checks.
         
@@ -1080,12 +1077,11 @@ def _quality_assurance_node(self, state: MessagesState):
         logger.info(f"✅ QualityAssuranceAgent checking quality...")
 
         # Call QualityAssuranceAgent
-        import asyncio
         task = {
             "query": query,
             "context": context
         }
-        qa_result = asyncio.run(self.quality_assurance.execute(task))
+        qa_result = await self.quality_assurance.execute(task)
 
         # Extract quality metrics
         qa_data = qa_result.get("output_data", {})
@@ -1170,7 +1166,7 @@ async def _writer_node(self, state: MessagesState):
         logger.info(f"✅ Answer generated: {len(answer)} chars")
         return {"messages": [AIMessage(content=answer)]}
 
-    def _citation_verification_node(self, state: MessagesState):
+    async def _citation_verification_node(self, state: MessagesState):
         """
         Citation Verification Node - Verify citations are accurate and relevant.
         
@@ -1231,7 +1227,7 @@ def _citation_verification_node(self, state: MessagesState):
 - Recommendations for improvement"""
 
         try:
-            verification_response = llm.invoke([HumanMessage(content=verification_prompt)])
+            verification_response = await llm.ainvoke([HumanMessage(content=verification_prompt)])
             verification_analysis = verification_response.content
 
             # Store verification results as system message
@@ -2055,7 +2051,7 @@ def _route_after_tools(self, state: MessagesState) -> Literal["generate_answer",
                 logger.info("⚠️ FALLBACK: Routing to generate_answer (grading failed + human_in_loop=False)")
                 return "generate_answer"
 
-    def _rewrite_question(self, state: MessagesState):
+    async def _rewrite_question(self, state: MessagesState):
         """
         Rewrite the original question for better retrieval.
         
@@ -2086,7 +2082,7 @@ def _rewrite_question(self, state: MessagesState):
 
         # Format prompt with original question
         prompt = f"{rewrite_prompt_template}\n\nOriginal question: {question}\n\nFormulate an improved question:"
-        response = self.llm.invoke([HumanMessage(content=prompt)])
+        response = await self.llm.ainvoke([HumanMessage(content=prompt)])
 
         logger.info(f"📝 Rewrote question: '{question}' → '{response.content}'")
 

apps/rag_management_app.py

@@ -2409,7 +2409,7 @@ def system_settings_page():
     if langsmith_enabled:
         st.success("✅ LangSmith Tracing Enabled")
         st.info(f"**Project:** {os.environ.get('LANGCHAIN_PROJECT', 'default')}")
-        st.markdown("🔗 View traces at: [https://smith.langchain.com/](https://smith.langchain.com/)")
+        st.markdown("🔗 View traces at: =[https://smith.langchain.com/](https://smith.langchain.com/)")
     else:
         st.warning("⚠️ LangSmith Tracing Disabled")
         st.info("To enable tracing, add to `.streamlit/secrets.toml`:")

context/context_engine.py

@@ -175,11 +175,15 @@ def _initialize_semantic_search(self) -> None:
 
             # Initialize Qdrant client (local, embedded - no API key needed)
             # Initialize this even if embeddings failed, so we can still manage collections
-            qdrant_path = Path(self.config.vector_db_path) / "qdrant_storage"
-            qdrant_path.mkdir(parents=True, exist_ok=True)
-            
-            self.qdrant_client = QdrantClient(path=str(qdrant_path))
-            self.logger.info(f"✅ Qdrant initialized (local storage: {qdrant_path})")
+            if self.config.vector_db_path == ":memory:":
+                self.qdrant_client = QdrantClient(location=":memory:")
+                self.logger.info("✅ Qdrant initialized (in-memory)")
+            else:
+                qdrant_path = Path(self.config.vector_db_path) / "qdrant_storage"
+                qdrant_path.mkdir(parents=True, exist_ok=True)
+                
+                self.qdrant_client = QdrantClient(path=str(qdrant_path))
+                self.logger.info(f"✅ Qdrant initialized (local storage: {qdrant_path})")
 
             if self.embeddings is None:
                 self.logger.warning("⚠️ No embedding provider available - semantic search disabled")

langgraph.json

@@ -4,7 +4,8 @@
   ],
   "graphs": {
     "agile_factory": "agents/agile_factory/workflow.py:graph",
-    "code_generator_test": "agents/agile_factory/test_code_generator_studio.py:graph"
+    "rag_agent": "agents/rag/rag_swarm_coordinator.py:graph",
+    "research_assistant": "agents/research_assistant/research_assistant.py:graph"
   },
   "env": {
     "GOOGLE_API_KEY": "${GOOGLE_API_KEY}",
@@ -13,7 +14,7 @@
     "LANGCHAIN_TRACING_V2": "true",
     "LANGCHAIN_PROJECT": "ai-dev-agent",
     "PYTHONPATH": ".",
-    "DATABASE_URI": "sqlite://:memory:",
+    "DATABASE_URI": "sqlite:///:memory:",
     "REDIS_URI": "redis://localhost:6379"
   },
   "python_version": "3.11",

prompts/langsmith_cache/agentic_rag_system.backup.txt

@@ -0,0 +1 @@
+prompt=PromptTemplate(input_variables=[], input_types={}, partial_variables={}, template='You are an intelligent AI assistant with advanced conversation memory, document retrieval, and quality analysis capabilities.\n\nCONVERSATION AWARENESS (CRITICAL):\n- ALWAYS review the full conversation history before making decisions\n- Track what information you\'ve already provided in this conversation\n- Build upon previous answers instead of repeating the same information\n- Reference earlier messages when relevant\n- If asked for "something new" or "more", provide information NOT mentioned before\n- Acknowledge follow-up questions that reference previous context\n- Remember what topics have been discussed and what questions have been asked\n\nINTELLIGENT WORKFLOW:\n1. Review conversation history to understand full context\n2. Call retrieve_project_docs to search for relevant information\n3. Grade retrieved documents for relevance using your document grading capability\n4. If documents are highly relevant: Proceed to answer generation\n5. If documents are not relevant: Rewrite the query for better results and retry\n6. Generate a comprehensive answer using the best retrieved context\n\nDOCUMENT GRADING:\n- Evaluate each retrieved document for relevance to the user\'s question\n- Consider both direct relevance and contextual relevance based on conversation history\n- Use grading results to determine whether to answer or refine the query\n\nQUERY REWRITING:\n- If documents are not relevant, reformulate the query for better retrieval\n- Consider conversation context when rewriting queries\n- Make queries more specific based on what the user is actually asking\n\nRESPONSE GENERATION:\n- Use the retrieved context and conversation history to generate answers\n- Be concise and well-structured\n- Cite sources when appropriate\n- If you\'ve already answered a similar question, acknowledge it and provide NEW details\n- Never repeat the exact same information - always add value\n\nMEMORY UTILIZATION:\n- Remember previous questions: "Your last question was about X"\n- Remember your previous responses: "I previously explained X, now let me add Y"\n- Track conversation flow: "Building on what we discussed about X..."\n- Avoid repetition: Always check if you\'ve already provided this information\n\nIMPORTANT: You have access to the full conversation history and intelligent grading/rewriting capabilities. Use all these tools to provide context-aware, high-quality, non-repetitive responses.\n') additional_kwargs={}