# prompting/zero_shot.py from dataclasses import dataclass from typing import Optional @dataclass class PromptTemplate: system: str user: str variables: list[str] def format(self, **kwargs) -> dict: return { "system": self.system, "user": self.user.format(**kwargs) } # Basic templates SUMMARIZE = PromptTemplate( system="You are a concise summarizer.", user="Summarize the following text in {num_sentences} sentences:\n\n{text}", variables=["num_sentences", "text"] ) EXTRACT_ENTITIES = PromptTemplate( system="You extract structured entities from text.", user="Extract {entity_types} from this text:\n\n{text}\n\nOutput as JSON.", variables=["entity_types", "text"] ) CLASSIFY = PromptTemplate( system="You are a precise classifier.", user="Classify this text into ONE category: {categories}\n\nText: {text}\n\nCategory:", variables=["categories", "text"] ) 

# prompting/few_shot.py from typing import List, Dict def build_few_shot_prompt( task_description: str, examples: List[Dict[str, str]], test_input: str, output_format: str = "" ) -> str: """Build a few-shot prompt with consistent formatting.""" parts = [task_description, ""] for i, ex in enumerate(examples, 1): parts.append(f"Example {i}:") parts.append(f"Input: {ex['input']}") parts.append(f"Output: {ex['output']}") parts.append("") parts.append(f"Input: {test_input}") if output_format: parts.append(f"Output ({output_format}):") else: parts.append("Output:") return "\n".join(parts) # Example: SQL generation SQL_EXAMPLES = [ { "input": "Find all users who signed up last month", "output": "SELECT * FROM users WHERE signup_date >= date('now', '-1 month')" }, { "input": "Count orders per customer", "output": "SELECT customer_id, COUNT(*) as order_count FROM orders GROUP BY customer_id" }, { "input": "Get products with price > 100", "output": "SELECT * FROM products WHERE price > 100" } ] prompt = build_few_shot_prompt( task_description="Convert natural language to SQLite SQL.", examples=SQL_EXAMPLES, test_input="Show me the top 5 customers by total spend", output_format="SQL only, no markdown" ) 

# prompting/cot.py COT_PROMPT = """Solve this problem step by step. Problem: {problem} Think through this carefully: 1. First, identify what is being asked 2. Break down the problem into sub-problems 3. Solve each sub-problem 4. Combine results for final answer Show your reasoning at each step. Step 1:""" COT_PROMPT_STRUCTURED = """{problem} Let's think step by step. {scratchpad_start}""" # Self-consistency: multiple CoT paths, majority vote async def self_consistency_cot(llm, problem: str, n: int = 5) -> str: """Run CoT multiple times, return majority answer.""" import asyncio from collections import Counter async def single_cot(): response = await llm.ainvoke(COT_PROMPT.format(problem=problem)) # Extract final answer (last number or conclusion)  return extract_answer(response) results = await asyncio.gather(*[single_cot() for _ in range(n)]) counter = Counter(results) return counter.most_common(1)[0][0] def extract_answer(text: str) -> str: # Extract final answer from CoT output  import re # Look for "Answer: X" or final number  answer_match = re.search(r"(?:Answer|Final answer|Result):\s*(.+)", text, re.IGNORECASE) if answer_match: return answer_match.group(1).strip() # Fallback: last sentence  return text.strip().split('\n')[-1] 

# prompting/tot.py from dataclasses import dataclass from typing import List, Optional import asyncio @dataclass class Thought: content: str score: float = 0.0 parent: Optional['Thought'] = None children: List['Thought'] = None def __post_init__(self): if self.children is None: self.children = [] class TreeOfThoughts: def __init__(self, llm, branching_factor: int = 3, max_depth: int = 3): self.llm = llm self.branching_factor = branching_factor self.max_depth = max_depth async def solve(self, problem: str) -> str: # Generate initial thoughts  root = Thought(content=problem) current_level = [root] for depth in range(self.max_depth): next_level = [] for thought in current_level: # Generate branches  branches = await self._generate_branches(thought.content) for branch in branches[:self.branching_factor]: child = Thought(content=branch, parent=thought) thought.children.append(child) next_level.append(child) # Evaluate and prune  scored = await self._evaluate_thoughts(next_level, problem) scored.sort(key=lambda x: x.score, reverse=True) current_level = scored[:self.branching_factor] # Get best path  best = current_level[0] path = [] while best: path.append(best.content) best = best.parent return " → ".join(reversed(path)) async def _generate_branches(self, thought: str) -> List[str]: prompt = f"""Given this partial solution: {thought} Generate {self.branching_factor} different next steps or approaches. Each should be a distinct reasoning direction. Return as numbered list.""" response = await self.llm.ainvoke(prompt) # Parse numbered list  return [line.split('. ', 1)[1] for line in response.split('\n') if '. ' in line] async def _evaluate_thoughts(self, thoughts: List[Thought], problem: str) -> List[Thought]: for thought in thoughts: prompt = f"""Problem: {problem} Proposed solution path: {thought.content} Rate the promise of this approach on a scale of 0-10. Consider: correctness, completeness, feasibility. Return only a number.""" response = await self.llm.ainvoke(prompt) try: thought.score = float(response.strip()) except: thought.score = 0.0 return thoughts 

# prompting/react.py REACT_TEMPLATE = """You are an agent that can use tools to solve problems. Available tools: {tools} Format: Question: {question} Thought: I need to... Action: tool_name Action Input: {{"arg": "value"}} Observation: [tool result] ... (repeat as needed) Thought: I now know the answer Final Answer: {answer} Begin! Question: {question} Thought:""" class ReActPromptBuilder: def __init__(self, tools: list[dict]): self.tools = tools def build(self, question: str) -> str: tool_desc = "\n".join(f"- {t['name']}: {t['description']}" for t in self.tools) return REACT_TEMPLATE.format(tools=tool_desc, question=question) def parse_response(self, response: str) -> dict: import re # Check for final answer  final_match = re.search(r"Final Answer:\s*(.+)", response, re.DOTALL) if final_match: return {"type": "final", "content": final_match.group(1).strip()} # Check for action  action_match = re.search(r"Action:\s*(\w+)", response) input_match = re.search(r"Action Input:\s*(\{.+\})", response) if action_match and input_match: import json return { "type": "action", "tool": action_match.group(1), "args": json.loads(input_match.group(1)) } return {"type": "thought", "content": response} 

# prompting/rag_prompts.py RAG_PROMPT = """You are a precise assistant that answers questions using ONLY the provided context. ## Context Documents {context} ## Instructions 1. Answer ONLY from the provided context 2. If context is insufficient, say "I don't have enough information to answer this" 3. Cite sources inline using [doc_N] format where N is the document number 4. Be concise but complete 5. If asked for a list, use bullet points ## Question {question} ## Answer:""" RAG_PROMPT_STRUCTURED = """{context} ## Question {question} ## Output Format (JSON) {{ "answer": "Your answer with [doc_N] citations", "citations": [ {{"doc_id": "doc_1", "quote": "exact quote from doc", "relevance": "high"}} ], "confidence": 0.95, "insufficient_context": false }}""" def format_rag_context(docs: list, max_chars: int = 8000) -> str: """Format retrieved documents for RAG prompt.""" parts = [] total = 0 for i, doc in enumerate(docs, 1): content = doc.get('content', '')[:2000] # Limit per doc  source = doc.get('metadata', {}).get('source', 'Unknown') part = f"[doc_{i}] Source: {source}\n{content}" if total + len(part) > max_chars: break parts.append(part) total += len(part) return "\n\n---\n\n".join(parts) 

# prompting/agent_prompts.py SUPERVISOR_PROMPT = """You are a Supervisor Agent managing a team of specialist agents. ## Your Role - Decompose complex user requests into subtasks - Delegate subtasks to appropriate specialists - Synthesize specialist outputs into coherent responses - Track progress and handle failures ## Specialist Agents Available {specialists} ## Delegation Protocol When delegating, use this format: DELEGATE: {{"agent": "agent_name", "task": "specific task description", "context": {{}}}} ## Response Format 1. If task is complete: Provide final synthesized answer 2. If delegation needed: Output DELEGATE command only 3. If clarification needed: Ask specific question ## Current Request {request} ## Previous Delegation Results {history} Your response:""" ### Specialist Agent (Researcher)  

 ### Code Generation Agent 

 ## Part 4: Prompt Optimization ### Automated Prompt Optimization 

async def optimize( self, base_prompt: str, test_cases: List[dict], # {"input": "", "expected": ""} mutation_rate: float = 0.3 ) -> PromptCandidate: # Initialize population population = await self._initialize_population(base_prompt) for gen in range(self.generations): # Evaluate scored = await self._evaluate_population(population, test_cases) scored.sort(key=lambda x: x.score, reverse=True) # Select top performers elite = scored[:self.population_size // 2] # Generate next generation population = elite.copy() while len(population) < self.population_size: parent = elite[0] # Best parent if random.random() < mutation_rate: child = await self._mutate(parent.prompt) else: child = await self._crossover(elite[0].prompt, elite[1].prompt) population.append(PromptCandidate(prompt=child)) return scored[0] async def _initialize_population(self, base: str) -> List[PromptCandidate]: population = [PromptCandidate(prompt=base)] # Generate variations variations = [ "Add more explicit instructions", "Add few-shot examples", "Add chain-of-thought prompting", "Add output format specification", "Add role/persona definition", "Add constraint specifications", "Add step-by-step reasoning", "Add self-verification step", "Add edge case handling", "Add few-shot with diverse examples" ] for var in variations[:self.population_size - 1]: prompt = await self._apply_variation(base, var) population.append(PromptCandidate(prompt=prompt)) return population async def _mutate(self, prompt: str) -> str: mutations = [ "Add 'Think step by step' to the prompt", "Add 'Be concise' instruction", "Add output format specification", "Add 'Verify your answer before responding'", "Add role definition at the start", "Add constraint: no markdown unless asked", "Add 'If uncertain, say so explicitly'", "Restructure into clear sections" ] return await self._apply_variation(prompt, random.choice(mutations)) async def _crossover(self, p1: str, p2: str) -> str: # Combine best parts of two prompts prompt = f"""Combine the best elements of these two prompts: 

async def _apply_variation(self, prompt: str, variation: str) -> str: prompt_text = f"""Improve this prompt by: {variation} 

 ### DSPy-Style Optimization