""" Example 2: Workflow & Agent Logs -> Skill Extraction This example demonstrates how to extract skills from workflow descriptions and agent runtime logs, then output them to a Markdown file. Usage: export OPENAI_API_KEY=your_api_key python examples/example_2_skill_extraction.py """ import asyncio import os import sys from openai import AsyncOpenAI from memu.app import MemoryService # Add src to sys.path src_path = os.path.abspath("src") sys.path.insert(0, src_path) async def generate_skill_md( all_skills, service, output_file, attempt_number, total_attempts, categories=None, is_final=False ): """ Use LLM to generate a concise task execution guide (skill.md). This creates a production-ready guide incorporating lessons learned from deployment attempts. """ os.makedirs(os.path.dirname(output_file), exist_ok=True) # Prepare context for LLM skills_text = "\n\n".join([f"### From {skill_data['source']}\n{skill_data['skill']}" for skill_data in all_skills]) # Get category summaries if available categories_text = "" if categories: categories_with_content = [cat for cat in categories if cat.get("summary") and cat.get("summary").strip()] if categories_with_content: categories_text = "\n\n".join([ f"**{cat.get('name', 'unknown')}**:\n{cat.get('summary', '')}" for cat in categories_with_content ]) # Construct prompt for LLM prompt = f"""Generate a concise production-ready task execution guide. **Context**: - Task: Production Microservice Deployment with Blue-Green Strategy - Progress: {attempt_number}/{total_attempts} attempts - Status: {"Complete" if is_final else f"v0.{attempt_number}"} **Skills Learned**: {skills_text} {f"**Categories**:\n{categories_text}" if categories_text else ""} **Required Structure**: 1. **Frontmatter** (YAML): - name: production-microservice-deployment - description: Brief description - version: {"1.0.0" if is_final else f"0.{attempt_number}.0"} - status: {"Production-Ready" if is_final else "Evolving"} 2. **Introduction**: What this guide does and when to use it 3. **Deployment Context**: Strategy, environment, goals 4. **Pre-Deployment Checklist**: - Actionable checks from lessons learned - Group by category (Database, Monitoring, etc.) - Mark critical items 5. **Deployment Procedure**: - Step-by-step instructions with commands - Include monitoring points 6. **Rollback Procedure**: - When to rollback (thresholds) - Exact commands - Expected recovery time 7. **Common Pitfalls & Solutions**: - Failures/issues encountered - Root cause, symptoms, solution 8. **Best Practices**: - What works well - Expected timelines 9. **Key Takeaways**: 3-5 most important lessons **Style**: - Use markdown with clear hierarchy - Be specific and concise - Technical and production-grade tone - Focus on PRACTICAL steps **CRITICAL**: - ONLY use information from provided skills/lessons - DO NOT make assumptions or add generic advice - Extract ACTUAL experiences from the logs Generate the complete markdown document now:""" client = AsyncOpenAI(api_key=service.llm_config.api_key) response = await client.chat.completions.create( model=service.llm_config.chat_model, messages=[ { "role": "system", "content": "You are an expert technical writer creating concise, production-grade deployment guides from real experiences.", }, {"role": "user", "content": prompt}, ], temperature=0.7, max_tokens=3000, ) generated_content = response.choices[0].message.content # Write to file with open(output_file, "w", encoding="utf-8") as f: f.write(generated_content) return True async def main(): """ Extract skills from agent logs using incremental memory updates. This example demonstrates INCREMENTAL LEARNING: 1. Process files ONE BY ONE 2. Each file UPDATES existing memory 3. Category summaries EVOLVE with each new file 4. Final output shows accumulated knowledge """ print("Example 2: Incremental Skill Extraction") print("-" * 50) # Get OpenAI API key from environment api_key = os.getenv("OPENAI_API_KEY") if not api_key: msg = "Please set OPENAI_API_KEY environment variable" raise ValueError(msg) # Custom config for skill extraction skill_prompt = """ You are analyzing an agent execution log. Extract the key actions taken, their outcomes, and lessons learned. For each significant action or phase: 1. **Action/Phase**: What was being attempted? 2. **Status**: SUCCESS ✅ or FAILURE ❌ 3. **What Happened**: What was executed 4. **Outcome**: What worked/failed, metrics 5. **Root Cause** (for failures): Why did it fail? 6. **Lesson**: What did we learn? 7. **Action Items**: Concrete steps for next time **IMPORTANT**: - Focus on ACTIONS and outcomes - Be specific: include actual metrics, errors, timing - ONLY extract information explicitly stated - DO NOT infer or assume information Extract ALL significant actions from the text: Text: {resource} """ # Define custom categories skill_categories = [ {"name": "deployment_execution", "description": "Deployment actions, traffic shifting, environment management"}, { "name": "pre_deployment_validation", "description": "Capacity validation, configuration checks, readiness verification", }, { "name": "incident_response_rollback", "description": "Incident response, error detection, rollback procedures", }, { "name": "performance_monitoring", "description": "Metrics monitoring, performance analysis, bottleneck detection", }, {"name": "database_management", "description": "Database capacity planning, optimization, schema changes"}, {"name": "testing_verification", "description": "Testing, smoke tests, load tests, verification"}, {"name": "infrastructure_setup", "description": "Kubernetes, containers, networking configuration"}, {"name": "lessons_learned", "description": "Key reflections, root cause analyses, action items"}, ] memorize_config = { "memory_types": ["skill"], "memory_type_prompts": {"skill": skill_prompt}, "memory_categories": skill_categories, } # Initialize service with OpenAI using llm_profiles # The "default" profile is required and used as the primary LLM configuration service = MemoryService( llm_profiles={ "default": { "api_key": api_key, "chat_model": "gpt-4o-mini", }, }, memorize_config=memorize_config, ) # Resources to process resources = [ ("examples/resources/logs/log1.txt", "document"), ("examples/resources/logs/log2.txt", "document"), ("examples/resources/logs/log3.txt", "document"), ] # Process each resource sequentially print("\nProcessing files...") all_skills = [] categories = [] for idx, (resource_file, modality) in enumerate(resources, 1): if not os.path.exists(resource_file): continue try: result = await service.memorize(resource_url=resource_file, modality=modality) # Extract skill items for item in result.get("items", []): if item.get("memory_type") == "skill": all_skills.append({"skill": item.get("summary", ""), "source": os.path.basename(resource_file)}) # Categories are returned in the result and updated after each memorize call categories = result.get("categories", []) # Generate intermediate skill.md await generate_skill_md( all_skills=all_skills, service=service, output_file=f"examples/output/skill_example/log_{idx}.md", attempt_number=idx, total_attempts=len(resources), categories=categories, ) except Exception as e: print(f"Error: {e}") # Generate final comprehensive skill.md await generate_skill_md( all_skills=all_skills, service=service, output_file="examples/output/skill_example/skill.md", attempt_number=len(resources), total_attempts=len(resources), categories=categories, is_final=True, ) print(f"\n✓ Processed {len(resources)} files, extracted {len(all_skills)} skills") print(f"✓ Generated {len(categories)} categories") print("✓ Output: examples/output/skill_example/") if __name__ == "__main__": asyncio.run(main())