1SYSTEM_PROMPT = """You are a research assistant.
2 
3CAPABILITIES:
4- Search the web for information
5- Read and analyze documents
6- Take notes and organize findings
7- Generate comprehensive reports
8 
9GUIDELINES:
10- Always cite your sources
11- Verify facts from multiple sources
12- Acknowledge uncertainty
13- Ask for clarification when needed
14 
15LIMITATIONS:
16- Cannot access private or paywalled content
17- Cannot make changes to external systems
18- Must respect rate limits on searches
19"""

1PROMPT = """
2Analyze customer feedback and extract themes.
3 
4EXAMPLE INPUT:
5"The app is fast but crashes when I try to export. Also wish it had dark mode."
6 
7EXAMPLE OUTPUT:
8{
9  "sentiment": "mixed",
10  "themes": [
11    {"topic": "performance", "sentiment": "positive", "detail": "app speed"},
12    {"topic": "stability", "sentiment": "negative", "detail": "export crashes"},
13    {"topic": "features", "sentiment": "neutral", "detail": "dark mode request"}
14  ]
15}
16 
17NOW ANALYZE:
18"{user_input}"
19"""

1# Good: specific about when to use
2{
3    "name": "search_knowledge_base",
4    "description": """Search company knowledge base.
5 
6    USE FOR:
7    - Company policies
8    - Product documentation
9    - How-to questions
10 
11    DO NOT USE FOR:
12    - General knowledge
13    - External information
14    - Real-time data
15 
16    Returns: List of relevant articles with titles and snippets"""
17}
18 
19# Bad: vague
20{
21    "name": "search",
22    "description": "Search for information"
23}

1TOOL_GUIDE = """
2Available tools and when to use them:
3 
41. **web_search**: Current events, general knowledge, external info
52. **knowledge_base**: Company-specific info, policies, products
63. **database_query**: User data, analytics, specific records
74. **calculator**: Any mathematical computation
8 
9Decision tree:
10- About our company? → knowledge_base
11- About user data? → database_query
12- Needs math? → calculator
13- Otherwise → web_search
14"""

1RESEARCH_TOOLS = [
2    {"name": "search_sources", "description": "Find relevant sources"},
3    {"name": "read_source", "description": "Extract info from a source"},
4    {"name": "take_note", "description": "Save a note with attribution"},
5    {"name": "get_notes", "description": "Retrieve all notes"},
6    {"name": "write_report", "description": "Generate report from notes"}
7]

1STRUCTURED_PROMPT = """
2Return your analysis in this exact JSON format:
3 
4{
5  "summary": "One paragraph overview",
6  "key_findings": [
7    {"finding": "string", "confidence": "high|medium|low", "evidence": "string"}
8  ],
9  "recommendations": ["string"],
10  "limitations": ["string"]
11}
12 
13Return ONLY the JSON, no additional text.
14"""

1from pydantic import BaseModel, validator
2from typing import List, Literal
3import json
4 
5class Finding(BaseModel):
6    finding: str
7    confidence: Literal["high", "medium", "low"]
8    evidence: str
9 
10class Analysis(BaseModel):
11    summary: str
12    key_findings: List[Finding]
13    recommendations: List[str]
14    limitations: List[str]
15 
16    @validator('key_findings')
17    def needs_findings(cls, v):
18        if len(v) < 1:
19            raise ValueError("Must have at least one finding")
20        return v
21 
22def parse_output(response: str) -> Analysis:
23    try:
24        data = json.loads(response)
25        return Analysis(**data)
26    except json.JSONDecodeError:
27        raise ValueError("Invalid JSON")
28    except Exception as e:
29        raise ValueError(f"Invalid format: {e}")

1async def stream_response(task: str):
2    async with client.messages.stream(
3        model=MODEL,
4        messages=[{"role": "user", "content": task}],
5        tools=TOOLS
6    ) as stream:
7        async for event in stream:
8            if event.type == "content_block_delta":
9                if event.delta.type == "text_delta":
10                    yield {"type": "text", "content": event.delta.text}
11 
12            elif event.type == "content_block_start":
13                if event.content_block.type == "tool_use":
14                    yield {"type": "tool_start", "tool": event.content_block.name}

1# 1. Stronger system prompt
2SYSTEM = """You MUST use tools to gather information.
3Do NOT rely on training data for facts."""
4 
5# 2. Explicit task instructions
6TASK = """
7Research {topic}.
8 
9IMPORTANT: Use web_search for EVERY fact you include.
10Do not include any information that doesn't come from a tool.
11"""
12 
13# 3. Require citations
14"""After each fact, cite the tool call that provided it."""

1seen_calls = set()
2 
3for call in tool_calls:
4    key = (call.name, json.dumps(call.input, sort_keys=True))
5 
6    if key in seen_calls:
7        messages.append({
8            "role": "user",
9            "content": "You already tried this. Try something different or provide your answer."
10        })
11    else:
12        seen_calls.add(key)
13        # Execute tool...
14 
15# Also: hard limit on iterations
16if iteration_count >= MAX_ITERATIONS:
17    break

1# Use explicit markers
2RESULT_FORMAT = """
3[TOOL RESULT: {tool_name}]
4{result}
5[END TOOL RESULT]
6 
7Only use information within TOOL RESULT markers.
8"""
9 
10# Validate citations reference actual results
11def validate_citations(response: str, tool_results: list) -> bool:
12    # Check all cited facts appear in tool_results
13    pass

1import structlog
2 
3logger = structlog.get_logger()
4 
5def debug_run(task: str):
6    log = logger.bind(task_id=str(uuid4()))
7    log.debug("agent_start", task=task)
8 
9    for i, step in enumerate(steps):
10        log.debug("step", num=i, tool=step.tool, input=step.input)
11        result = execute(step)
12        log.debug("result", num=i, result=result[:200])
13 
14    log.debug("agent_done", total_steps=len(steps))

1def print_trace(trace: list):
2    for i, step in enumerate(trace):
3        print(f"""
4Step {i + 1}:
5  Thinking: {step.thinking[:100]}...
6  Tool: {step.tool}
7  Input: {json.dumps(step.input, indent=2)}
8  Result: {step.result[:100]}...
9  Duration: {step.duration:.2f}s
10""")

1COT_PROMPT = """
2Solve this step by step:
3 
4{problem}
5 
6Process:
71. Understand what's being asked
82. Identify what information you need
93. Gather information using tools
104. Analyze the information
115. Form your conclusion
126. Verify your answer
13 
14Show your reasoning at each step.
15"""

1SELF_CRITIQUE = """
2After completing the task, critique your work:
3 
41. Did you fully address the question?
52. Are your sources reliable and cited?
63. Are there gaps in your analysis?
74. What could be improved?
8 
9Based on your critique, revise if needed.
10"""

1REFLECTION = """
2The previous attempt had issues:
3{error_description}
4 
5Before trying again:
61. What went wrong?
72. Why did it happen?
83. How will you avoid it this time?
9 
10Now try again with these learnings.
11"""

1def iterative_refine(task: str, max_rounds: int = 3) -> str:
2    output = agent.run(task)
3 
4    for _ in range(max_rounds):
5        critique_prompt = f"""
6        Task: {task}
7 
8        Attempt:
9        {output}
10 
11        Critique this and provide an improved version.
12        """
13 
14        response = agent.run(critique_prompt)
15 
16        if "no improvements needed" in response.lower():
17            break
18 
19        output = extract_improved_version(response)
20 
21    return output

1async def parallel_explore(task: str, approaches: int = 3) -> str:
2    # Generate different approaches
3    approach_prompt = f"Generate {approaches} different ways to solve: {task}"
4    approaches = await agent.run(approach_prompt)
5 
6    # Execute in parallel
7    results = await asyncio.gather(*[
8        agent.run(f"Execute this approach: {a}")
9        for a in parse_approaches(approaches)
10    ])
11 
12    # Select best
13    selection_prompt = f"""
14    Task: {task}
15 
16    Results from {len(results)} approaches:
17    {format_results(results)}
18 
19    Select the best and explain why.
20    """
21    return await agent.run(selection_prompt)

1SENSITIVE_ACTIONS = ["delete_file", "send_email", "make_payment"]
2 
3def execute_with_approval(tool: str, params: dict) -> str:
4    if tool in SENSITIVE_ACTIONS:
5        print(f"Agent wants to: {tool}")
6        print(f"Parameters: {params}")
7        approved = input("Approve? (y/n): ")
8 
9        if approved.lower() != 'y':
10            return "Action not approved by user."
11 
12    return execute_tool(tool, params)

1def research(question: str) -> str:
2    return agent.run(f"""
3    Research: {question}
4 
5    Process:
6    1. Search for 3-5 authoritative sources
7    2. Read and extract key information
8    3. Identify agreement and disagreement
9    4. Synthesize into comprehensive answer
10 
11    Requirements:
12    - Cite every factual claim
13    - Note confidence levels
14    - Include diverse perspectives
15    - Acknowledge limitations
16 
17    Format: Summary → Detailed findings → Sources
18    """)

1def generate_code(spec: str) -> str:
2    return agent.run(f"""
3    Generate code for: {spec}
4 
5    Process:
6    1. Clarify requirements
7    2. Design approach
8    3. Write code with comments
9    4. Add error handling
10    5. Write example usage
11 
12    Output:
13    - Full implementation
14    - Approach explanation
15    - Example usage
16    - Dependencies/limitations
17    """)

1def analyze_data(question: str, context: str) -> str:
2    return agent.run(f"""
3    Analyze data to answer: {question}
4 
5    Context: {context}
6 
7    Process:
8    1. Understand question and available data
9    2. Write and execute analysis
10    3. Interpret results
11    4. Generate visualizations if helpful
12    5. Form conclusions
13 
14    Output:
15    - Direct answer
16    - Supporting analysis
17    - Visualizations
18    - Caveats
19    """)