| import asyncio |
| from llama_index.core.workflow import Context |
| from llama_index.core.agent.workflow import AgentWorkflow |
| from tools.tavily_search_tool import ( |
| search_web, record_notes, write_report, review_report, |
| SearchWebArgs, RecordNotesArgs, WriteReportArgs, ReviewReportArgs |
| ) |
|
|
| async def test_tools(): |
| """Test all tools with the new Pydantic signatures.""" |
| print("Testing tools with new Pydantic model arguments...") |
| |
| |
| workflow = AgentWorkflow(agents=[], root_agent=None) |
| ctx = Context(workflow) |
| |
| |
| await ctx.set("state", { |
| "research_notes": {}, |
| "report_content": "Not written yet.", |
| "review": "Review required.", |
| }) |
| |
| print("\n1. Testing search_web...") |
| try: |
| search_args = SearchWebArgs(query="artificial intelligence history") |
| search_result = await search_web(search_args) |
| print(f"β
search_web worked! Result length: {len(search_result)}") |
| print(f"Preview: {search_result[:200]}...") |
| except Exception as e: |
| print(f"β search_web failed: {e}") |
| |
| print("\n2. Testing record_notes...") |
| try: |
| notes_args = RecordNotesArgs( |
| notes="Test research notes about AI history", |
| notes_title="AI History Overview" |
| ) |
| notes_result = await record_notes(ctx, notes_args) |
| print(f"β
record_notes worked! Result: {notes_result}") |
| |
| |
| state = await ctx.get("state") |
| print(f"State after notes: {list(state.keys())}") |
| print(f"Research notes: {state.get('research_notes', {})}") |
| except Exception as e: |
| print(f"β record_notes failed: {e}") |
| |
| print("\n3. Testing write_report...") |
| try: |
| report_args = WriteReportArgs( |
| report_content="""# Artificial Intelligence History |
| |
| ## Introduction |
| This is a test report about AI history. |
| |
| ## Early Development |
| AI began in the 1950s with researchers like Alan Turing. |
| |
| ## Modern Era |
| Today, AI includes machine learning and deep learning. |
| |
| ## Conclusion |
| AI continues to evolve rapidly.""", |
| title="Test AI History Report" |
| ) |
| report_result = await write_report(ctx, report_args) |
| print(f"β
write_report worked! Result: {report_result}") |
| |
| |
| state = await ctx.get("state") |
| print(f"Report content length: {len(state.get('report_content', ''))}") |
| print(f"Has structured report: {'structured_report' in state}") |
| except Exception as e: |
| print(f"β write_report failed: {e}") |
| |
| print("\n4. Testing review_report...") |
| try: |
| review_args = ReviewReportArgs(review="APPROVED: The report looks good!") |
| review_result = await review_report(ctx, review_args) |
| print(f"β
review_report worked! Result: {review_result}") |
| |
| |
| state = await ctx.get("state") |
| print(f"Final review: {state.get('review', 'No review')}") |
| except Exception as e: |
| print(f"β review_report failed: {e}") |
| |
| print("\n5. Final state check...") |
| final_state = await ctx.get("state") |
| print(f"Final state keys: {list(final_state.keys())}") |
| print(f"Research notes count: {len(final_state.get('research_notes', {}))}") |
| print(f"Report written: {final_state.get('report_content', 'Not written') != 'Not written yet.'}") |
|
|
| if __name__ == "__main__": |
| asyncio.run(test_tools()) |