The Final Step from Theory to Practice
The previous four tutorials covered the basics of Agent Loop, tool calling, Reflection, and multi-agent collaboration. Now it's time to put it all together—we will build a complete Agent Loop application that can automatically handle the entire process from requirement understanding to final delivery.
This application will simulate an "AI Research Assistant": the user proposes a research topic, and the Agent automatically performs requirement analysis, task decomposition, information retrieval, result verification, and report generation. The entire process requires no human intervention, showcasing the full capabilities of Agent Loop in real-world scenarios.
Application Architecture Design
Our "AI Research Assistant" adopts a five-stage pipeline architecture:
- Requirement Reception: Understand the user's research topic, clarify scope and depth requirements
- Task Decomposition: Break down the research topic into multiple executable subtasks
- Step-by-Step Execution: Complete subtasks one by one, collecting information and analysis results
- Result Verification: Cross-verify the accuracy of information, supplement missing content
- Final Delivery: Integrate all results to generate a structured research report
Each stage may trigger feedback loops—if verification fails, it returns to the execution stage to re-collect information; if delivery is incomplete, it returns to the decomposition stage for re-planning.
Skill Chain Breakdown: Complete Agent Loop Application
Using the "Complete Agent Loop Application" skill chain as an example, we break down the role of each node in the end-to-end process:
Node 1: Requirement Reception (sp-15)—This is the entry point of the entire Agent Loop. The AI needs to extract key information from the user's natural language input: research topic, expected depth, output format, special requirements, etc. The output of this node is a structured requirements document.
Node 2: Task Decomposition (sp-188)—Transform the requirements document into an executable action plan. Good task decomposition should have: each subtask independently executable, clear logical relationships between subtasks, and reasonable and controllable total workload.
Node 3: Step-by-Step Execution (sp-187)—This is the core execution phase of the Agent Loop. The AI follows the task plan and completes each subtask one by one. Each subtask may involve tool calls, information retrieval, code execution, etc. Execution logs are recorded for later verification.
Node 4: Result Verification (sp-185)—Perform quality checks on all execution results. Verification dimensions include: information accuracy (whether there are factual errors), coverage completeness (whether key points are missed), and logical consistency (whether parts contradict each other). If verification fails, specific issues are fed back to the execution node for correction.
Node 5: Final Delivery (sp-122)—Integrate all verified results into the final deliverable. Generate output that meets the user's required format (report, PPT, code repository, etc.). This node also adds metadata (generation time, information sources, confidence scores, etc.).
Complete Code Implementation
Below is the complete implementation code for the "AI Research Assistant":
import json import time from datetime import datetime from openai import OpenAI client = OpenAI( api_key="your-deepseek-api-key", base_url="https://api.deepseek.com" ) class AIResearchAssistant: """Complete Agent Loop application: AI Research Assistant""" def __init__(self, max_iterations=10): self.max_iterations = max_iterations self.execution_log = [] self.start_time = None def log(self, phase, message): """Record execution log""" entry = { "timestamp": datetime.now().isoformat(), "phase": phase, "message": message } self.execution_log.append(entry) print(f"[{phase}] {message}") def call_llm(self, system_prompt, user_prompt, response_format=None): """Unified LLM call interface""" messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ] kwargs = { "model": "deepseek-chat", "messages": messages } if response_format: kwargs["response_format"] = response_format response = client.chat.completions.create(**kwargs) return response.choices[0].message.content def phase1_understand(self, user_input): """Phase 1: Requirement reception and analysis""" self.log("Requirement Analysis", "Start analyzing user requirements") prompt = f"""Analyze the following user research requirements and extract key information. User input: {user_input} Please output in JSON format: {{ "topic": "Research topic", "scope": "Research scope (technical/business/academic, etc.)", "depth": "Expected depth (overview/detailed/in-depth)", "format": "Expected output format", "keywords": ["keyword1", "keyword2"], "constraints": ["constraint1", "constraint2"], "target_audience": "Target audience" }}""" result = self.call_llm( "You are a requirements analysis expert, skilled at extracting structured requirements from vague descriptions.", prompt ) requirements = json.loads(result) self.log("Requirement Analysis", f"Identified topic: {requirements['topic']}") return requirements def phase2_decompose(self, requirements): """Phase 2: Task Decomposition""" self.log("Task Decomposition", "Starting to decompose research tasks") prompt = f"""Based on the following requirements, decompose the research task into 3-5 executable subtasks. Requirements: {json.dumps(requirements, ensure_ascii=False)} Please output in JSON format: {{ "subtasks": [ {{ "id": 1, "title": "Subtask title", "description": "Detailed description", "type": "research|analysis|synthesis|verification", "estimated_tokens": 500, "depends_on": [] }} ], "execution_order": [1, 2, 3] }}""" result = self.call_llm( "You are a project management expert skilled at breaking down complex tasks into executable subtasks.", prompt ) plan = json.loads(result) self.log("Task Decomposition", f"Decomposed into {len(plan['subtasks'])} subtasks") return plan def phase3_execute(self, requirements, plan): """Phase 3: Step-by-Step Execution""" self.log("Execution Phase", "Starting to execute subtasks") results = {} for task_id in plan["execution_order"]: task = next(t for t in plan["subtasks"] if t["id"] == task_id) self.log("Execution Phase", f"Executing subtask {task_id}: {task['title']}") # Collect context from dependent tasks context = "" for dep_id in task.get("depends_on", []): if dep_id in results: context += f"\nResult of prerequisite task {dep_id}:\n{results[dep_id]}" prompt = f"""Complete the following research subtask. Research topic: {requirements['topic']} Subtask: {task['title']} Description: {task['description']} {"Context: " + context if context else ""} Requirements: 1. Provide accurate, in-depth information 2. Cite specific data and cases 3. Point out uncertainties in information (if any) 4. Indicate the source type of information (academic paper/industry report/news/official data, etc.)""" result = self.call_llm( "You are a senior researcher skilled in collecting, analyzing, and synthesizing information.", prompt ) results[task_id] = result self.log("Execution Phase", f"Subtask {task_id} completed ({len(result)} characters)") return results def phase4_verify(self, requirements, results): """Phase 4: Result Verification""" self.log("Verification Phase", "Starting to verify result quality") prompt = f"""Verify the completeness and accuracy of the following research results. Research topic: {requirements['topic']} Research results: {json.dumps(results, ensure_ascii=False, indent=2)} Evaluate from the following dimensions: 1. Information accuracy (any dubious or inconsistent content) 2. Coverage completeness (any key aspects missing) 3. Logical consistency (any contradictions between parts) Output in JSON format: {{ "verified": true/false, "overall_score": 0-100, "issues": [ {{"task_id": 1, "severity": "high/medium/low", "description": "Issue description"}} ], "missing_topics": ["Missing topics"], "recommendation": "Pass/Needs supplement/Needs redo" }}""" result = self.call_llm( "You are a rigorous quality audit expert who does not overlook any information flaws.", prompt ) verification = json.loads(result) if verification["verified"]: self.log("Verification Phase", f"Verification passed, score: {verification['overall_score']}") else: self.log("Verification Phase", f"Found {len(verification['issues'])} issues") return verification def phase5_deliver(self, requirements, results, verification): """Phase 5: Final Delivery""" self.log("Delivery Phase", "Generating final report") elapsed = time.time() - self.start_time prompt = f"""Based on the following research results, generate the final research report. Research topic: {requirements['topic']} Target audience: {requirements['target_audience']} Expected format: {requirements['format']} Verification results: {json.dumps(verification, ensure_ascii=False)} Research data: {json.dumps(results, ensure_ascii=False, indent=2)} Report requirements: 1. Clear structure, including abstract, body, and conclusion 2. Professional yet understandable, suitable for the target audience 3. Mark key data and citation sources 4. At the end, include execution statistics (elapsed time, number of subtasks, quality score) Execution statistics: - Total elapsed: {elapsed:.1f} seconds - Number of subtasks: {len(results)} - Quality score: {verification['overall_score']}""" report = self.call_llm( "You are a senior report writing expert, skilled at transforming research data into high-quality reports.", prompt ) self.log("Delivery Phase", f"Report generation completed ({len(report)} characters)") return report def run(self, user_input): """Run the complete Agent Loop""" self.start_time = time.time() print("=" * 60) print("AI Research Assistant Started") print("=" * 60) for iteration in range(self.max_iterations): print(f"\n--- Iteration {iteration + 1} ---") # Phases 1-2: Requirement Analysis and Task Decomposition if iteration == 0: requirements = self.phase1_understand(user_input) plan = self.phase2_decompose(requirements) # Phase 3: Execution results = self.phase3_execute(requirements, plan) # Phase 4: Verification verification = self.phase4_verify(requirements, results) if verification["verified"]: # Phase 5: Delivery report = self.phase5_deliver(requirements, results, verification) print("\n" + "=" * 60) print("Task Completed!") elapsed = time.time() - self.start_time print(f"Total elapsed: {elapsed:.1f} seconds") print(f"Total iterations: {iteration + 1}") print(f"Number of subtasks: {len(results)}") print(f"Quality score: {verification['overall_score']}") print("=" * 60) return { "report": report, "requirements": requirements, "verification": verification, "execution_log": self.execution_log, "elapsed_seconds": elapsed } else: self.log("Iteration", "Verification failed, adjusting plan and re-executing") # Adjust plan based on verification feedback plan = self.phase2_decompose({ **requirements, "feedback": verification["issues"], "missing": verification["missing_topics"] }) return {"error": "Maximum iterations reached, task incomplete"} # Usage example assistant = AIResearchAssistant(max_iterations=10) result = assistant.run( "Please research the current status and future trends of AI Agent technology in 2024, " "focusing on enterprise-level application scenarios, targeting technical decision-makers, " "and output a structured research report." ) print(f"\nResearch Report:\n{result['report']}") re> Running Effects and Key Metrics
Running the above code, you will see the complete execution process of the Agent Loop:
- Requirement Analysis: Extract information such as research topic, scope, depth, and target audience
- Task Decomposition: Break down the research into subtasks such as background investigation, technical analysis, application scenarios, competitive landscape, and trend prediction
- Step-by-Step Execution: Each subtask executes independently, outputting detailed analysis results
- Result Verification: Check information accuracy, coverage completeness, and logical consistency
- Final Delivery: Generate a structured research report, including execution statistics
Key design points:
- Logging System: Record execution logs at each stage for debugging and auditing
- Feedback Loop: Automatically return to the task decomposition stage when verification fails, and re-plan
- Context Passing: Pass context between subtasks through dependencies to ensure information coherence
- Quality Quantification: Use a scoring mechanism to quantify output quality as a basis for stopping conditions
Production Deployment Recommendations
To deploy this Agent Loop application to a production environment, the following aspects need to be considered:
- Asynchronous Execution: Use message queues (such as Redis, RabbitMQ) for asynchronous task processing to avoid long blocking
- State Persistence: Store the Agent's execution state in a database to support resumable execution and progress queries
- Concurrency Control: Implement rate limiting and retry mechanisms for LLM API calls to avoid triggering throttling
- Cost Monitoring: Record token consumption for each call and set budget alerts
- Human Review Nodes: Add human review steps at key decision points (such as before final delivery)
- A/B Testing: Compare the effectiveness of different prompt strategies to continuously optimize Agent performance
Summary and Outlook
Through these five tutorials, we started from the basic concepts of Agent Loop, gradually delved into tool calling, Reflection, multi-Agent collaboration, and finally built a complete Agent Loop application. Agent Loop is not an unattainable concept but a development paradigm that can be adopted immediately.
The future of Agent Loop is exciting: autonomous programming, automated scientific research, intelligent operations—these scenarios are turning from imagination into reality. Mastering Agent Loop means mastering the core capability to build next-generation AI applications. Now, open the skill chain and orchestrate your first Agent Loop yourself!
Want to orchestrate this skill chain yourself?
Open in Skill Chain →