1import asyncio
2from textwrap import dedent
3
4from kern.agent import Agent
5from kern.knowledge.embedder.openai import OpenAIEmbedder
6from kern.knowledge.knowledge import Knowledge
7from kern.models.openai import OpenAIResponses
8from kern.tools.knowledge import KnowledgeTools
9from kern.vectordb.lancedb import LanceDb, SearchType
10
11# Create a knowledge containing information from a URL
12print("Setting up URL knowledge...")
13agno_docs = Knowledge(
14 # Use LanceDB as the vector database
15 vector_db=LanceDb(
16 uri="tmp/lancedb",
17 table_name="cookbook_knowledge_tools",
18 search_type=SearchType.hybrid,
19 embedder=OpenAIEmbedder(id="text-embedding-3-small"),
20 ),
21)
22# Add content to the knowledge
23asyncio.run(agno_docs.ainsert(url="https://www.paulgraham.com/read.html"))
24print("Knowledge ready.")
25
26
27print("\n=== Example 1: Using KnowledgeTools in non-streaming mode ===\n")
28
29# Create agent with KnowledgeTools
30agent = Agent(
31 model=OpenAIResponses(id="gpt-5.2"),
32 tools=[
33 KnowledgeTools(
34 knowledge=agno_docs,
35 think=True,
36 search=True,
37 analyze=True,
38 add_instructions=True,
39 )
40 ],
41 instructions=dedent("""\
42 You are an expert problem-solving assistant with strong analytical skills! ��
43 Use the knowledge tools to organize your thoughts, search for information,
44 and analyze results step-by-step.
45 \
46 """),
47 markdown=True,
48)
49
50# Run the agent (non-streaming) using agent.run() to get the response
51print("Running with KnowledgeTools (non-streaming)...")
52response = agent.run(
53 "What does Paul Graham explain here with respect to need to read?", stream=False
54)
55
56# Check reasoning_content from the response
57print("\n--- reasoning_content from response ---")
58if hasattr(response, "reasoning_content") and response.reasoning_content:
59 print("✅ reasoning_content FOUND in non-streaming response")
60 print(f" Length: {len(response.reasoning_content)} characters")
61 print("\n=== reasoning_content preview (non-streaming) ===")
62 preview = response.reasoning_content[:1000]
63 if len(response.reasoning_content) > 1000:
64 preview += "..."
65 print(preview)
66else:
67 print("❌ reasoning_content NOT FOUND in non-streaming response")
68
69
70print("\n\n=== Example 2: Using KnowledgeTools in streaming mode ===\n")
71
72# Create a fresh agent for streaming
73streaming_agent = Agent(
74 model=OpenAIResponses(id="gpt-5.2"),
75 tools=[
76 KnowledgeTools(
77 knowledge=agno_docs,
78 think=True,
79 search=True,
80 analyze=True,
81 add_instructions=True,
82 )
83 ],
84 instructions=dedent("""\
85 You are an expert problem-solving assistant with strong analytical skills! ��
86 Use the knowledge tools to organize your thoughts, search for information,
87 and analyze results step-by-step.
88 \
89 """),
90 markdown=True,
91)
92
93# Process streaming responses and look for the final RunOutput
94print("Running with KnowledgeTools (streaming)...")
95final_response = None
96for event in streaming_agent.run(
97 "What does Paul Graham explain here with respect to need to read?",
98 stream=True,
99 stream_events=True,
100):
101 # Print content as it streams (optional)
102 if hasattr(event, "content") and event.content:
103 print(event.content, end="", flush=True)
104
105 # The final event in the stream should be a RunOutput object
106 if hasattr(event, "reasoning_content"):
107 final_response = event
108
109print("\n\n--- reasoning_content from final stream event ---")
110if (
111 final_response
112 and hasattr(final_response, "reasoning_content")
113 and final_response.reasoning_content
114):
115 print("✅ reasoning_content FOUND in final stream event")
116 print(f" Length: {len(final_response.reasoning_content)} characters")
117 print("\n=== reasoning_content preview (streaming) ===")
118 preview = final_response.reasoning_content[:1000]
119 if len(final_response.reasoning_content) > 1000:
120 preview += "..."
121 print(preview)
122else:
123 print("❌ reasoning_content NOT FOUND in final stream event")
