Spaces:
Runtime error
Runtime error
| import json | |
| import logging | |
| from app.db.database import SessionLocal | |
| from app.services.conversation_service import ConversationService | |
| from app.agents.dev_agent import DevAgent | |
| from app.db.models import DevSessionModel, ConversationMessageModel | |
| # Disable logging clutter in CLI mode | |
| logging.getLogger("litellm").setLevel(logging.WARNING) | |
| def main(): | |
| print("====================================================") | |
| print(" Universal Dev Agent Builder CLI ") | |
| print("====================================================") | |
| print("Describe the agent you want to build.") | |
| print("Type 'exit' or 'quit' to close the session.\n") | |
| db = SessionLocal() | |
| conv_service = ConversationService() | |
| agent = DevAgent() | |
| # 1. Create a persistent session in the database | |
| session_id = conv_service.create_session(db) | |
| try: | |
| while True: | |
| user = input("\nYou: ").strip() | |
| if not user: | |
| continue | |
| if user.lower() in ("exit", "quit"): | |
| print("Exiting builder. Goodbye!") | |
| break | |
| # 2. Save user message | |
| conv_service.add_message(db, session_id, 'user', user) | |
| # 3. Load full history and previous state | |
| session = db.query(DevSessionModel).filter(DevSessionModel.id == session_id).first() | |
| msgs = db.query(ConversationMessageModel).filter(ConversationMessageModel.session_id == session_id).order_by(ConversationMessageModel.created_at.asc()).all() | |
| history = [{"role": m.role, "content": m.content} for m in msgs] | |
| state_init = { | |
| 'extracted_requirements': session.extracted_requirements or {}, | |
| 'missing_information': session.missing_information or [], | |
| 'clarification_questions': session.clarification_questions or [], | |
| 'draft_spec': session.draft_spec, | |
| 'validation_result': session.validation_result, | |
| 'status': session.status | |
| } | |
| # 4. Invoke agent workflow | |
| result = agent.process(session_id, user, history, state_init) | |
| # 5. Save assistant response and update state in DB | |
| assistant_content = "" | |
| questions = result.get('clarification_questions', []) or [] | |
| if questions: | |
| assistant_content = "\n".join([q['question'] for q in questions]) | |
| elif result.get('draft_spec'): | |
| if result.get('status') == 'completed': | |
| assistant_content = "Here is the final generated Agent Spec!" | |
| else: | |
| assistant_content = "I have drafted the Agent Spec, but validation errors occurred." | |
| if assistant_content: | |
| conv_service.add_message(db, session_id, 'assistant', assistant_content) | |
| session.extracted_requirements = result.get('extracted_requirements', {}) | |
| session.missing_information = result.get('missing_information', []) | |
| session.clarification_questions = result.get('clarification_questions', []) | |
| session.draft_spec = result.get('draft_spec', None) | |
| session.validation_result = result.get('validation_result', None) | |
| session.status = result.get('status', 'collecting_requirements') | |
| db.commit() | |
| db.refresh(session) | |
| # 6. Display Agent questions or spec/errors | |
| if questions: | |
| for q in questions: | |
| print(f"\nAgent: {q['question']}") | |
| if session.draft_spec: | |
| print("\n--- Current Agent Spec Draft ---") | |
| print(json.dumps(session.draft_spec, indent=2)) | |
| print("--------------------------------") | |
| if session.validation_result: | |
| vr = session.validation_result | |
| if vr.get('valid'): | |
| print("\n[SUCCESS] The Agent Spec is fully VALID and generated!") | |
| print("You can copy the JSON above. Exiting...") | |
| break | |
| else: | |
| print("\n[WARNING] Validation errors detected in draft:") | |
| for err in vr.get('errors', []): | |
| print(f" - [{err.get('code')}] {err.get('message')} (path: {err.get('path')})") | |
| print("\nPlease address the validation errors above.") | |
| finally: | |
| db.close() | |
| if __name__ == "__main__": | |
| main() | |