1SYSTEM_PROMPT = """You are a helpful research assistant.
2 
3Your capabilities:
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"""

Research AI agents

1Research the current state of AI agents in customer support.
2 
3Focus on:
41. Major platforms and their features
52. Implementation challenges
63. Success metrics and case studies
74. Cost considerations
8 
9Output format:
10- Executive summary (3-5 sentences)
11- Detailed findings by topic
12- Recommendations
13- Sources cited
14 
15Target length: 1500-2000 words

1PROMPT_WITH_EXAMPLES = """
2Analyze customer feedback and extract key 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 this feedback:
18"{user_input}"
19"""

1{
2    "name": "search_knowledge_base",
3    "description": """Search the company knowledge base for information.
4 
5    Use this tool when:
6    - User asks about company policies
7    - User needs product documentation
8    - User asks 'how to' questions
9 
10    Do NOT use for:
11    - General knowledge questions
12    - External information
13    - Real-time data
14 
15    Returns: List of relevant articles with titles and snippets""",
16    "input_schema": {
17        "type": "object",
18        "properties": {
19            "query": {
20                "type": "string",
21                "description": "Search query - use specific keywords"
22            },
23            "category": {
24                "type": "string",
25                "enum": ["policies", "products", "procedures"],
26                "description": "Filter by category for better results"
27            }
28        },
29        "required": ["query"]
30    }
31}

1TOOL_SELECTION_PROMPT = """
2Available tools and when to use them:
3 
41. **web_search**: For current events, general knowledge, external information
52. **knowledge_base**: For company-specific information, policies, products
63. **database_query**: For user data, analytics, specific records
74. **calculator**: For any mathematical computations
8 
9Decision tree:
10- Is it about our company? -> knowledge_base
11- Is it about the user's data? -> database_query
12- Does it need math? -> calculator
13- Otherwise -> web_search
14"""

1# Tools that build on each other
2RESEARCH_TOOLS = [
3    {
4        "name": "search_sources",
5        "description": "Find relevant sources for a topic"
6    },
7    {
8        "name": "read_source",
9        "description": "Extract information from a specific source"
10    },
11    {
12        "name": "take_note",
13        "description": "Save a note with source attribution"
14    },
15    {
16        "name": "get_notes",
17        "description": "Retrieve all notes taken so far"
18    },
19    {
20        "name": "write_report",
21        "description": "Generate report from notes"
22    }
23]

1STRUCTURED_OUTPUT_PROMPT = """
2Analyze the provided data and return your findings 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 
13Important: Return ONLY the JSON, no additional text.
14"""

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

1async def stream_agent_response(task: str):
2    """Stream agent response with tool call handling."""
3 
4    async with client.messages.stream(
5        model="claude-sonnet-4-20250514",
6        max_tokens=4096,
7        messages=[{"role": "user", "content": task}],
8        tools=TOOLS
9    ) as stream:
10        async for event in stream:
11            if event.type == "content_block_delta":
12                if event.delta.type == "text_delta":
13                    yield {"type": "text", "content": event.delta.text}
14 
15            elif event.type == "content_block_start":
16                if event.content_block.type == "tool_use":
17                    yield {
18                        "type": "tool_start",
19                        "tool": event.content_block.name
20                    }

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

1# 1. Track and detect loops
2seen_calls = set()
3for call in tool_calls:
4    key = (call.name, json.dumps(call.input, sort_keys=True))
5    if key in seen_calls:
6        # Inject guidance
7        messages.append({
8            "role": "user",
9            "content": "You've already tried this. Try a different approach or provide your answer with available information."
10        })
11    seen_calls.add(key)
12 
13# 2. Limit iterations
14MAX_TOOL_CALLS = 10
15if len(tool_calls) >= MAX_TOOL_CALLS:
16    # Force completion
17    pass

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

1import structlog
2 
3logger = structlog.get_logger()
4 
5def debug_agent_run(task: str):
6    log = logger.bind(task_id=str(uuid4()))
7 
8    log.debug("starting_agent", task=task)
9 
10    for i, step in enumerate(agent_steps):
11        log.debug("agent_step",
12                  step=i,
13                  thinking=step.thinking,
14                  tool=step.tool,
15                  input=step.input)
16 
17        result = execute_tool(step.tool, step.input)
18 
19        log.debug("tool_result",
20                  step=i,
21                  tool=step.tool,
22                  result=result[:200])  # Truncate for logging
23 
24    log.debug("agent_complete", steps=len(agent_steps))

1def visualize_agent_trace(trace: list):
2    """Create visual representation of agent execution."""
3 
4    output = []
5    for i, step in enumerate(trace):
6        output.append(f"""
7Step {i + 1}:
8  Thinking: {step.thinking[:100]}...
9  Tool: {step.tool}
10  Input: {json.dumps(step.input, indent=2)}
11  Result: {step.result[:100]}...
12  Duration: {step.duration:.2f}s
13""")
14 
15    return "\n".join(output)

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

1SELF_CRITIQUE_PROMPT = """
2After completing the task, critique your own 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_PROMPT = """
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_refinement(task: str, max_iterations: int = 3) -> str:
2    """Refine agent output through multiple passes."""
3 
4    current_output = agent.run(task)
5 
6    for i in range(max_iterations):
7        # Ask for critique
8        critique_prompt = f"""
9        Here is an attempt at: {task}
10 
11        Attempt:
12        {current_output}
13 
14        Critique this attempt. Be specific about:
15        - What's good
16        - What's missing or wrong
17        - How to improve it
18 
19        Then provide an improved version.
20        """
21 
22        response = agent.run(critique_prompt)
23 
24        # Extract improved version
25        if "improved version" in response.lower():
26            current_output = extract_improved_version(response)
27        else:
28            break  # No improvements needed
29 
30    return current_output

1async def parallel_exploration(task: str, num_approaches: int = 3) -> str:
2    """Try multiple approaches in parallel and select best."""
3 
4    # Generate different approaches
5    approach_prompt = f"""
6    For this task: {task}
7 
8    Generate {num_approaches} different approaches to solving it.
9    Format as:
10    APPROACH 1: ...
11    APPROACH 2: ...
12    """
13 
14    approaches = await agent.run(approach_prompt)
15 
16    # Execute each approach in parallel
17    results = await asyncio.gather(*[
18        agent.run(f"Execute this approach: {approach}")
19        for approach in parse_approaches(approaches)
20    ])
21 
22    # Select best result
23    selection_prompt = f"""
24    Task: {task}
25 
26    Here are {len(results)} different results:
27    {format_results(results)}
28 
29    Select the best one and explain why.
30    """
31 
32    return await agent.run(selection_prompt)

1def run_with_approval(task: str, checkpoints: list) -> str:
2    """Run agent with human approval at key points."""
3 
4    messages = [{"role": "user", "content": task}]
5 
6    while True:
7        response = agent.step(messages)
8 
9        # Check if at a checkpoint
10        if should_checkpoint(response, checkpoints):
11            # Present to human
12            print(f"Agent wants to: {describe_action(response)}")
13            approved = input("Approve? (y/n): ")
14 
15            if approved.lower() != 'y':
16                # Add human guidance
17                guidance = input("Guidance: ")
18                messages.append({
19                    "role": "user",
20                    "content": f"Do not do that. Instead: {guidance}"
21                })
22                continue
23 
24        # Execute
25        messages = execute_step(response, messages)
26 
27        if response.stop_reason == "end_turn":
28            break
29 
30    return extract_final_response(messages)

1def research_question(question: str) -> str:
2    return agent.run(f"""
3    Research this question thoroughly: {question}
4 
5    Process:
6    1. Search for 3-5 authoritative sources
7    2. Read and extract key information from each
8    3. Identify areas of agreement and disagreement
9    4. Synthesize into a comprehensive answer
10 
11    Requirements:
12    - Cite every factual claim
13    - Note confidence levels
14    - Include diverse perspectives
15    - Acknowledge limitations
16 
17    Format: Start with summary, then detailed findings, then sources.
18    """)

1def generate_code(specification: str) -> str:
2    return agent.run(f"""
3    Generate code for: {specification}
4 
5    Process:
6    1. Clarify requirements (ask if needed)
7    2. Design the solution approach
8    3. Write the code with comments
9    4. Add error handling
10    5. Write example usage
11 
12    Output:
13    - Code block with full implementation
14    - Brief explanation of approach
15    - Example usage
16    - Notes on dependencies or limitations
17    """)

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