ede5577 · ollama_semantic_function_calling

Changed files (16) +284 -3437

ollamatools-project/CHANGELOG.md +14 -0

…	…	@@ -5,6 +5,20 @@ All notable changes to the OllamaTools project will be documented in this file.
5	5	The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
6	6	and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
7	7
	8	## [1.0.0] - 2025-04-20
	9
	10	### Breaking Changes
	11	- Removed the TransparentOllamaAgent implementation completely
	12	- Removed the entire ollamatools/agent/transparent directory
	13	- Updated examples to use ModularOllamaAgent instead
	14	- Updated tests to use ModularOllamaAgent instead
	15	- Updated documentation to reflect the removal of TransparentOllamaAgent
	16
	17	### Changed
	18	- ModularOllamaAgent is now the primary agent implementation
	19	- Simplified codebase by focusing on a single agent implementation
	20	- Updated migration guide to help users transition to ModularOllamaAgent
	21
8	22	## [0.2.0] - 2025-04-15
9	23
10	24	### Added

ollamatools-project/docs/migration_guide.md +6 -18

…	…	@@ -1,10 +1,10 @@
1		# Migrating from TransparentOllamaAgent to ModularOllamaAgent
	1	# Migration to ModularOllamaAgent
2	2
3		This guide provides instructions for migrating from the TransparentOllamaAgent to the new ModularOllamaAgent.
	3	This guide provides instructions for migrating to the ModularOllamaAgent.
4	4
5	5	## Overview
6	6
7		The ModularOllamaAgent is a drop-in replacement for the TransparentOllamaAgent, with the same API but a more modular and maintainable implementation. This guide will help you migrate your code to use the new ModularOllamaAgent.
	7	The ModularOllamaAgent is the primary agent implementation in OllamaTools, with a modular and maintainable architecture. This guide will help you migrate your code to use the ModularOllamaAgent if you were previously using the now-removed TransparentOllamaAgent.
8	8
9	9	## Basic Migration
10	10
…	…	@@ -174,13 +174,11 @@ class CustomAgent(ModularOllamaAgent):
174	174	pass
175	175	```
176	176
177		## Backward Compatibility
	177	## Breaking Changes
178	178
179		For backward compatibility, the TransparentOllamaAgent class is still available and will continue to work. However, it is now an alias for ModularOllamaAgent and will show a deprecation warning:
	179	The TransparentOllamaAgent class and the entire transparent agent implementation have been removed from the codebase. If your code was using the TransparentOllamaAgent, you must update it to use the ModularOllamaAgent instead.
180	180
181		```
182		DeprecationWarning: TransparentOllamaAgent is deprecated and will be removed in a future version. Use ModularOllamaAgent from ollamatools.agent.modular instead.
183		```
	181	This is a breaking change that was made to simplify the codebase and focus development efforts on the modular implementation.
184	182
185	183	## New Features in ModularOllamaAgent
186	184
…	…	@@ -240,16 +238,6 @@ from ollamatools.agent.modular import print_status, print_success, print_error
240	238
241	239	If you encounter "method not found" errors, check if you're using methods that have been renamed or moved. Refer to the [ModularOllamaAgent Guide](modular_agent_guide.md) for a complete list of available methods.
242	240
243		### Deprecation Warnings
244
245		If you continue to use the TransparentOllamaAgent class, you'll see deprecation warnings. To suppress these warnings, you can use:
246
247		```python
248		import warnings
249		warnings.filterwarnings("ignore", category=DeprecationWarning)
250		```
251
252		However, it's recommended to migrate to the ModularOllamaAgent class instead.
253	241
254	242	## Getting Help
255	243

ollamatools-project/examples/README_MODULAR_AGENT.md +235 -0

…	…	@@ -0,0 +1,235 @@
	1	# Modular Ollama Agent Examples
	2
	3	This directory contains examples of the Modular Ollama Agent:
	4
	5	1. `improved_transparent_agent.py` - An implementation using the ModularOllamaAgent
	6	2. `extended_transparent_agent.py` - An example of extending the ModularOllamaAgent with custom functionality
	7
	8	## Comparison of Implementations
	9
	10	### Original Implementation (`transparent_ollama_agent.py`)
	11
	12	The original implementation:
	13	- Contains all functionality in a single file (~1000+ lines)
	14	- Defines the `TransparentOllamaAgent` class with all methods directly in the class
	15	- Handles search, command management, issue tracking, and other functionality in the same file
	16	- Is more difficult to maintain and extend due to its monolithic nature
	17
	18	### Improved Implementation (`improved_transparent_agent.py`)
	19
	20	The improved implementation:
	21	- Uses the modular agent code from `ollamatools/agent/modular/`
	22	- Is much shorter (~150 lines) because it leverages the modular components
	23	- Imports the `ModularOllamaAgent` class and utility functions from the modular code
	24	- Has comprehensive functionality with better code organization
	25
	26	## Benefits of the Modular Approach
	27
	28	1. Better Code Organization: Functionality is logically separated into different modules:
	29	- `agent.py`: Main agent implementation
	30	- `search.py`: Search functionality
	31	- `commands.py`: Command management
	32	- `issues.py`: Issue tracking
	33	- `utils.py`: Utility functions
	34	- `feedback.py`: Semantic function selection feedback
	35
	36	2. Easier Maintenance: Changes to one aspect (e.g., search) don't require modifying the entire agent
	37
	38	3. Improved Reusability: Components can be reused in other agents or scripts
	39
	40	4. Cleaner Implementation: The main agent class delegates to specialized modules
	41
	42	5. Better Testing: Modules can be tested independently
	43
	44	## How to Use the Modular Agent
	45
	46	To use the modular agent in your own scripts:
	47
	48	```python
	49	from ollamatools.agent import ModularOllamaAgent
	50	from ollamatools.agent.modular import (
	51	print_status, print_success, print_info
	52	)
	53
	54	# Initialize the agent
	55	agent = ModularOllamaAgent(
	56	model="llama3.1:latest",
	57	permission_prompts=True
	58	)
	59
	60	# Use the agent
	61	response = agent.chat("Hello, how can you help me?")
	62	print(response)
	63	```
	64
	65	## Extending the Modular Agent
	66
	67	If you want to extend the functionality of the modular agent:
	68
	69	1. Create a new module in `ollamatools/agent/` for your specific functionality
	70	2. Import and use the existing modules as needed
	71	3. Create a new agent class that inherits from `ModularOllamaAgent` if necessary
	72
	73	Example of extending the agent:
	74
	75	```python
	76	from ollamatools.agent import ModularOllamaAgent
	77
	78	class EnhancedAgent(ModularOllamaAgent):
	79	"""An enhanced version of the ModularOllamaAgent with additional capabilities."""
	80
	81	def __init__(self, args, *kwargs):
	82	super().__init__(args, *kwargs)
	83	# Add your custom initialization here
	84
	85	def new_feature(self):
	86	"""Implement a new feature."""
	87	# Your implementation here
	88	pass
	89	```
	90
	91	## Running the Examples
	92
	93	To run the improved implementation:
	94	```
	95	python improved_transparent_agent.py
	96	```
	97
	98	To run the extended implementation with custom features:
	99	```
	100	python extended_transparent_agent.py
	101	```
	102
	103	The extended implementation demonstrates how to add custom features by inheriting from the modular agent.
	104
	105	## Semantic Function Selection Feedback
	106
	107	The improved agent now includes a feedback system that enhances its ability to understand natural language queries and map them to appropriate functions. This system:
	108
	109	1. Learns from User Interactions: Records when functions are used correctly or incorrectly
	110	2. Improves Over Time: Gets better at selecting the right function for similar queries
	111	3. Adapts to User Language: Learns the specific ways users phrase their requests
	112	4. Provides Analytics: Offers insights into common patterns and mismatches
	113
	114	### How the Feedback System Works
	115
	116	1. Data Collection: The system records:
	117	- Which queries led to which function calls
	118	- Whether the function calls were successful
	119	- User corrections when the wrong function was used
	120	- Similarity scores between current and past queries
	121
	122	2. Pattern Recognition: The system analyzes:
	123	- Common words and phrases in similar queries
	124	- Which functions work best for different types of questions
	125	- Patterns of mismatches between expected and actual functions
	126	- Implicit feedback from execution results
	127
	128	3. Improved Selection: When a new query comes in:
	129	- The system compares it to previous similar queries
	130	- It suggests the function that worked best for similar queries
	131	- It calculates confidence scores for function suggestions
	132	- It applies this knowledge before falling back to default behavior
	133
	134	### Technical Implementation
	135
	136	The feedback system is implemented in the `ollamatools/agent/modular/feedback.py` module through the `FunctionSelectionFeedback` class:
	137
	138	1. Feedback Storage:
	139	- Feedback is stored in a JSONL file at `~/.ollamatools/function_selection_feedback.jsonl`
	140	- Recent feedback is cached in memory for quick access
	141	- Each feedback entry includes query, function information, timestamp, and similarity data
	142
	143	2. Similarity Calculation:
	144	- Uses Jaccard similarity between query word sets
	145	- Compares new queries against previously recorded queries
	146	- Filters out common stop words to focus on meaningful terms
	147	- Returns top similar queries with their similarity scores
	148
	149	3. Function Suggestion:
	150	- Weights function suggestions by similarity scores
	151	- Calculates confidence based on consistency of function usage
	152	- Returns the best function match with confidence score
	153
	154	4. Pattern Generation:
	155	- Automatically generates regex patterns from feedback data
	156	- Creates patterns based on common words in similar queries
	157	- Identifies consistent query patterns that map to specific functions
	158	- Suggests improvements for function selection logic
	159
	160	5. Command Intent Detection:
	161	- Specialized support for detecting system command intents
	162	- Maps natural language questions to appropriate shell commands
	163	- Uses both predefined patterns and learned patterns from feedback
	164	- Formats command results in a user-friendly way
	165
	166	### Using the Feedback System
	167
	168	The improved agent includes commands for interacting with the feedback system:
	169
	170	- `feedback <query> \| <correct_function>` - Provide explicit feedback on function selection
	171	- `analyze_feedback` - Show analysis of function selection patterns and improvement suggestions
	172
	173	Example:
	174	```
	175	You: what directory am I in
	176	Assistant: [uses read_file incorrectly]
	177
	178	You: feedback 'what directory am I in' \| execute_command
	179	Assistant: Thank you for your feedback! I'll remember to use execute_command for similar queries in the future.
	180
	181	[Later]
	182	You: where am I located?
	183	Assistant: [now correctly uses pwd command]
	184	```
	185
	186	### Advanced Usage
	187
	188	Developers can leverage the feedback system in their own agents:
	189
	190	```python
	191	from ollamatools.agent.modular.feedback import FunctionSelectionFeedback
	192
	193	# Initialize the feedback system
	194	feedback_system = FunctionSelectionFeedback()
	195
	196	# Get function suggestions for a query
	197	function, confidence, similar_queries = feedback_system.get_function_suggestion("what files are in this folder?")
	198
	199	# Record explicit feedback
	200	feedback_system.record_explicit_feedback(
	201	query="what files are in this folder?",
	202	intended_function="execute_command",
	203	actual_function="read_file",
	204	user_correction=True
	205	)
	206
	207	# Record implicit feedback from execution results
	208	feedback_system.record_implicit_feedback(
	209	query="show me the files",
	210	function="execute_command",
	211	confidence_score=0.8,
	212	execution_result={"returncode": 0, "stdout": "file1.txt file2.txt", "stderr": ""}
	213	)
	214
	215	# Generate improved patterns for function selection
	216	improved_patterns = feedback_system.generate_improved_patterns()
	217
	218	# Analyze function selection patterns
	219	analysis = feedback_system.analyze_function_selection_patterns()
	220	```
	221
	222	## Extended Agent Features
	223
	224	The `extended_transparent_agent.py` example demonstrates:
	225
	226	1. Conversation History Tracking: Saves all conversations to a JSON file
	227	2. Conversation Summarization: Provides summaries of recent conversations
	228	3. Topic Analysis: Analyzes conversation topics based on keywords
	229	4. Performance Metrics: Tracks response times and estimated token usage
	230
	231	This example shows how to:
	232	- Inherit from the `ModularOllamaAgent` class
	233	- Override methods to add custom functionality
	234	- Add new methods for additional features
	235	- Maintain compatibility with the base agent's functionality

ollamatools-project/examples/README_TRANSPARENT_AGENT.md +0 -241

…	…	@@ -1,241 +0,0 @@
1		# Transparent Ollama Agent Examples
2
3		This directory contains three implementations of the Transparent Ollama Agent:
4
5		1. `transparent_ollama_agent.py` - The original monolithic implementation
6		2. `improved_transparent_agent.py` - A new implementation using the modularized agent code
7		3. `extended_transparent_agent.py` - An example of extending the modularized agent with custom functionality
8
9		## Comparison of Implementations
10
11		### Original Implementation (`transparent_ollama_agent.py`)
12
13		The original implementation:
14		- Contains all functionality in a single file (~1000+ lines)
15		- Defines the `TransparentOllamaAgent` class with all methods directly in the class
16		- Handles search, command management, issue tracking, and other functionality in the same file
17		- Is more difficult to maintain and extend due to its monolithic nature
18
19		### Improved Implementation (`improved_transparent_agent.py`)
20
21		The improved implementation:
22		- Uses the modularized agent code from `ollamatools/agent/transparent/`
23		- Is much shorter (~150 lines) because it leverages the modular components
24		- Imports the `TransparentOllamaAgent` class and utility functions from the modularized code
25		- Has the same functionality but with better code organization
26
27		## Benefits of the Modularized Approach
28
29		1. Better Code Organization: Functionality is logically separated into different modules:
30		- `agent.py`: Main agent implementation
31		- `search.py`: Search functionality
32		- `commands.py`: Command management
33		- `issues.py`: Issue tracking
34		- `utils.py`: Utility functions
35		- `feedback.py`: Semantic function selection feedback
36
37		2. Easier Maintenance: Changes to one aspect (e.g., search) don't require modifying the entire agent
38
39		3. Improved Reusability: Components can be reused in other agents or scripts
40
41		4. Cleaner Implementation: The main agent class delegates to specialized modules
42
43		5. Better Testing: Modules can be tested independently
44
45		## How to Use the Modularized Agent
46
47		To use the modularized agent in your own scripts:
48
49		```python
50		from ollamatools.agent.transparent import (
51		TransparentOllamaAgent,
52		print_status, print_success, print_info
53		)
54
55		# Initialize the agent
56		agent = TransparentOllamaAgent(
57		model="llama3.1:latest",
58		permission_prompts=True
59		)
60
61		# Use the agent
62		response = agent.chat("Hello, how can you help me?")
63		print(response)
64		```
65
66		## Extending the Modularized Agent
67
68		If you want to extend the functionality of the modularized agent:
69
70		1. Create a new module in `ollamatools/agent/` for your specific functionality
71		2. Import and use the existing modules as needed
72		3. Create a new agent class that inherits from `TransparentOllamaAgent` if necessary
73
74		Example of extending the agent:
75
76		```python
77		from ollamatools.agent.transparent import TransparentOllamaAgent
78
79		class EnhancedAgent(TransparentOllamaAgent):
80		"""An enhanced version of the TransparentOllamaAgent with additional capabilities."""
81
82		def __init__(self, args, *kwargs):
83		super().__init__(args, *kwargs)
84		# Add your custom initialization here
85
86		def new_feature(self):
87		"""Implement a new feature."""
88		# Your implementation here
89		pass
90		```
91
92		## Running the Examples
93
94		To run the original implementation:
95		```
96		python transparent_ollama_agent.py
97		```
98
99		To run the improved implementation:
100		```
101		python improved_transparent_agent.py
102		```
103
104		To run the extended implementation with custom features:
105		```
106		python extended_transparent_agent.py
107		```
108
109		The original and improved implementations provide the same functionality, but the improved version uses the modularized code structure for better maintainability and extensibility. The extended implementation demonstrates how to add custom features by inheriting from the modularized agent.
110
111		## Semantic Function Selection Feedback
112
113		The improved agent now includes a feedback system that enhances its ability to understand natural language queries and map them to appropriate functions. This system:
114
115		1. Learns from User Interactions: Records when functions are used correctly or incorrectly
116		2. Improves Over Time: Gets better at selecting the right function for similar queries
117		3. Adapts to User Language: Learns the specific ways users phrase their requests
118		4. Provides Analytics: Offers insights into common patterns and mismatches
119
120		### How the Feedback System Works
121
122		1. Data Collection: The system records:
123		- Which queries led to which function calls
124		- Whether the function calls were successful
125		- User corrections when the wrong function was used
126		- Similarity scores between current and past queries
127
128		2. Pattern Recognition: The system analyzes:
129		- Common words and phrases in similar queries
130		- Which functions work best for different types of questions
131		- Patterns of mismatches between expected and actual functions
132		- Implicit feedback from execution results
133
134		3. Improved Selection: When a new query comes in:
135		- The system compares it to previous similar queries
136		- It suggests the function that worked best for similar queries
137		- It calculates confidence scores for function suggestions
138		- It applies this knowledge before falling back to default behavior
139
140		### Technical Implementation
141
142		The feedback system is implemented in the `ollamatools/agent/transparent/feedback.py` module through the `FunctionSelectionFeedback` class:
143
144		1. Feedback Storage:
145		- Feedback is stored in a JSONL file at `~/.ollamatools/function_selection_feedback.jsonl`
146		- Recent feedback is cached in memory for quick access
147		- Each feedback entry includes query, function information, timestamp, and similarity data
148
149		2. Similarity Calculation:
150		- Uses Jaccard similarity between query word sets
151		- Compares new queries against previously recorded queries
152		- Filters out common stop words to focus on meaningful terms
153		- Returns top similar queries with their similarity scores
154
155		3. Function Suggestion:
156		- Weights function suggestions by similarity scores
157		- Calculates confidence based on consistency of function usage
158		- Returns the best function match with confidence score
159
160		4. Pattern Generation:
161		- Automatically generates regex patterns from feedback data
162		- Creates patterns based on common words in similar queries
163		- Identifies consistent query patterns that map to specific functions
164		- Suggests improvements for function selection logic
165
166		5. Command Intent Detection:
167		- Specialized support for detecting system command intents
168		- Maps natural language questions to appropriate shell commands
169		- Uses both predefined patterns and learned patterns from feedback
170		- Formats command results in a user-friendly way
171
172		### Using the Feedback System
173
174		The improved agent includes commands for interacting with the feedback system:
175
176		- `feedback <query> \| <correct_function>` - Provide explicit feedback on function selection
177		- `analyze_feedback` - Show analysis of function selection patterns and improvement suggestions
178
179		Example:
180		```
181		You: what directory am I in
182		Assistant: [uses read_file incorrectly]
183
184		You: feedback 'what directory am I in' \| execute_command
185		Assistant: Thank you for your feedback! I'll remember to use execute_command for similar queries in the future.
186
187		[Later]
188		You: where am I located?
189		Assistant: [now correctly uses pwd command]
190		```
191
192		### Advanced Usage
193
194		Developers can leverage the feedback system in their own agents:
195
196		```python
197		from ollamatools.agent.transparent.feedback import FunctionSelectionFeedback
198
199		# Initialize the feedback system
200		feedback_system = FunctionSelectionFeedback()
201
202		# Get function suggestions for a query
203		function, confidence, similar_queries = feedback_system.get_function_suggestion("what files are in this folder?")
204
205		# Record explicit feedback
206		feedback_system.record_explicit_feedback(
207		query="what files are in this folder?",
208		intended_function="execute_command",
209		actual_function="read_file",
210		user_correction=True
211		)
212
213		# Record implicit feedback from execution results
214		feedback_system.record_implicit_feedback(
215		query="show me the files",
216		function="execute_command",
217		confidence_score=0.8,
218		execution_result={"returncode": 0, "stdout": "file1.txt file2.txt", "stderr": ""}
219		)
220
221		# Generate improved patterns for function selection
222		improved_patterns = feedback_system.generate_improved_patterns()
223
224		# Analyze function selection patterns
225		analysis = feedback_system.analyze_function_selection_patterns()
226		```
227
228		## Extended Agent Features
229
230		The `extended_transparent_agent.py` example demonstrates:
231
232		1. Conversation History Tracking: Saves all conversations to a JSON file
233		2. Conversation Summarization: Provides summaries of recent conversations
234		3. Topic Analysis: Analyzes conversation topics based on keywords
235		4. Performance Metrics: Tracks response times and estimated token usage
236
237		This example shows how to:
238		- Inherit from the `TransparentOllamaAgent` class
239		- Override methods to add custom functionality
240		- Add new methods for additional features
241		- Maintain compatibility with the base agent's functionality

ollamatools-project/examples/extended_transparent_agent.py +11 -11

…	…	@@ -1,8 +1,8 @@
1	1	#!/usr/bin/env python3
2	2	"""
3		Extended Transparent Ollama Agent Example.
	3	Extended Modular Ollama Agent Example.
4	4
5		This script demonstrates how to extend the modularized TransparentOllamaAgent
	5	This script demonstrates how to extend the ModularOllamaAgent
6	6	with custom functionality.
7	7	"""
8	8
…	…	@@ -13,9 +13,9 @@ import time
13	13	import logging
14	14	from typing import Dict, Any, List, Optional
15	15
16		# Import the modularized agent
17		from ollamatools.agent.transparent import (
18		TransparentOllamaAgent,
	16	# Import the modular agent
	17	from ollamatools.agent import ModularOllamaAgent
	18	from ollamatools.agent.modular import (
19	19	print_status, print_success, print_info, print_error
20	20	)
21	21	from ollamatools.opensearch_web_tools import DEFAULT_OPENSEARCH_CONFIG
…	…	@@ -25,11 +25,11 @@ from ollamatools.utils.logger import setup_logger
25	25	# Set up logging
26	26	logger = setup_logger("extended_agent", level="INFO")
27	27
28		class ExtendedTransparentAgent(TransparentOllamaAgent):
	28	class ExtendedModularAgent(ModularOllamaAgent):
29	29	"""
30		Extended version of the TransparentOllamaAgent with additional features.
	30	Extended version of the ModularOllamaAgent with additional features.
31	31
32		This class demonstrates how to extend the modularized agent with custom
	32	This class demonstrates how to extend the modular agent with custom
33	33	functionality while leveraging the existing features.
34	34	"""
35	35
…	…	@@ -188,14 +188,14 @@ def main():
188	188	"""Run the extended transparent Ollama agent."""
189	189	try:
190	190	# Initialize the extended agent
191		logger.info("Initializing ExtendedTransparentAgent...")
192		agent = ExtendedTransparentAgent(
	191	logger.info("Initializing ExtendedModularAgent...")
	192	agent = ExtendedModularAgent(
193	193	model="llama3.1:latest",
194	194	permission_prompts=True
195	195	)
196	196
197	197	logger.info("Agent initialized. Enter 'exit' to quit.")
198		print("\nExtended Transparent Ollama Agent")
	198	print("\nExtended Modular Ollama Agent")
199	199	print("=================================")
200	200	print("Type 'exit' to quit")
201	201	print("Type 'commands' to see available commands")

ollamatools-project/examples/improved_transparent_agent.py +9 -9

…	…	@@ -1,8 +1,8 @@
1	1	#!/usr/bin/env python3
2	2	"""
3		Improved Transparent Ollama Agent using the modularized agent code.
	3	Improved Modular Ollama Agent Example.
4	4
5		This script demonstrates how to use the modularized TransparentOllamaAgent
	5	This script demonstrates how to use the ModularOllamaAgent
6	6	to create a transparent, feature-rich agent with search capabilities,
7	7	command management, issue tracking, and semantic function selection feedback.
8	8	"""
…	…	@@ -12,9 +12,9 @@ import sys
12	12	import logging
13	13	from typing import Dict, Any
14	14
15		# Import the modularized agent
16		from ollamatools.agent.transparent import (
17		TransparentOllamaAgent,
	15	# Import the modular agent
	16	from ollamatools.agent import ModularOllamaAgent
	17	from ollamatools.agent.modular import (
18	18	print_status, print_success, print_info,
19	19	FunctionSelectionFeedback
20	20	)
…	…	@@ -42,9 +42,9 @@ OPENSEARCH_CONFIG.update({
42	42	def main():
43	43	"""Run the improved transparent Ollama agent."""
44	44	try:
45		# Initialize the agent using the modularized implementation
46		logger.info("Initializing TransparentOllamaAgent...")
47		agent = TransparentOllamaAgent(
	45	# Initialize the agent using the modular implementation
	46	logger.info("Initializing ModularOllamaAgent...")
	47	agent = ModularOllamaAgent(
48	48	model="llama3.1:latest",
49	49	permission_prompts=True,
50	50	opensearch_config=OPENSEARCH_CONFIG,
…	…	@@ -52,7 +52,7 @@ def main():
52	52	)
53	53
54	54	logger.info("Agent initialized. Enter 'exit' to quit.")
55		print("\nImproved Transparent Ollama Agent")
	55	print("\nImproved Modular Ollama Agent")
56	56	print("================================")
57	57	print("Type 'exit' to quit")
58	58	print("Type 'commands' to see available commands")

ollamatools-project/examples/transparent_ollama_agent.py +0 -1119

…	…	@@ -1,1119 +0,0 @@
1		#!/usr/bin/env python3
2		"""
3		Transparent Ollama Agent with OpenSearchNG Integration.
4
5		This script demonstrates an improved OllamaAgent that:
6		1. Shows transparent execution of commands and functions
7		2. Integrates with a locally running OpenSearchNG instance
8		3. Provides detailed feedback on what's happening
9		4. Handles errors gracefully with suggestions for fixes
10
11		Usage:
12		python transparent_ollama_agent.py
13		"""
14
15		import os
16		import sys
17		import json
18		import logging
19		import time
20		from typing import Dict, Any, List, Optional, Set, Union, Tuple
21
22		# Add the parent directory to the Python path to import ollamatools
23		sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
24
25		from ollamatools import OllamaAgent, Config
26		from ollamatools.command_validator import DEFAULT_ALLOWLIST
27		from ollamatools.utils.logger import setup_logger
28		from ollamatools.errors import (
29		OllamaToolsError,
30		CommandError,
31		FileError,
32		WebError,
33		handle_exception
34		)
35
36		# Import the OpenSearchNG integration
37		from ollamatools.opensearch_web_tools import (
38		opensearch_query,
39		opensearch_fetch_url,
40		get_opensearch_info,
41		DEFAULT_OPENSEARCH_CONFIG
42		)
43
44		# Set up logging
45		logger = setup_logger("transparent_ollama_agent", level="INFO")
46
47		# Extend the default command allowlist with useful commands
48		EXTENDED_ALLOWLIST = DEFAULT_ALLOWLIST.union({
49		"python", "python3", "calc", "bc", "find", "grep", "awk", "sed",
50		"head", "tail", "more", "less", "cat", "echo", "cd", "pwd"
51		})
52
53		# Configuration for OpenSearchNG
54		OPENSEARCH_CONFIG = DEFAULT_OPENSEARCH_CONFIG.copy()
55		OPENSEARCH_CONFIG.update({
56		"base_url": "http://localhost:8081", # Updated to your container's port
57		"engines": ["google", "bing", "duckduckgo", "brave"],
58		"num_results": 5
59		})
60
61		class TransparentOllamaAgent:
62		"""
63		An enhanced OllamaAgent that provides transparent execution and better feedback.
64		"""
65
66		def __init__(
67		self,
68		model: str = "llama3.1:latest",
69		permission_prompts: bool = True,
70		opensearch_config: Optional[Dict[str, Any]] = None,
71		command_allowlist: Optional[Set[str]] = None
72		):
73		"""
74		Initialize the agent.
75
76		Args:
77		model: Ollama model to use
78		permission_prompts: Whether to display permission prompts
79		opensearch_config: OpenSearchNG configuration
80		command_allowlist: Custom command allowlist
81		"""
82		# Initialize OllamaAgent
83		self.agent = OllamaAgent(
84		model=model,
85		permission_prompts=permission_prompts
86		)
87
88		# Store configuration
89		self.opensearch_config = opensearch_config or OPENSEARCH_CONFIG
90		self.command_allowlist = command_allowlist or EXTENDED_ALLOWLIST
91
92		# Initialize state
93		self.context = []
94		self.execution_history = []
95
96		# Initialize issue tracking
97		self.issues = []
98		self.issue_id_counter = 1
99
100		# Path for saving issues
101		self.issues_file = os.path.join(
102		os.path.expanduser('~'),
103		'.ollamatools',
104		'issues.json'
105		)
106
107		# Create directory if it doesn't exist
108		os.makedirs(os.path.dirname(self.issues_file), exist_ok=True)
109
110		# Load existing issues if any
111		self._load_issues()
112
113		# Check if OpenSearchNG is available
114		try:
115		self._print_status("Checking OpenSearchNG availability...")
116		info = get_opensearch_info(config=self.opensearch_config)
117		self._print_success(f"OpenSearchNG running at {self.opensearch_config['base_url']}")
118
119		# Log available engines
120		if 'engines' in info and info['engines']:
121		self._print_info(f"Available search engines: {', '.join(info['engines'])}")
122		except Exception as e:
123		self._print_error(f"OpenSearchNG not available: {str(e)}")
124		self._print_warning("Falling back to regular search methods")
125		self._suggest_fix("Make sure your OpenSearchNG instance is running at " +
126		f"{self.opensearch_config['base_url']}")
127
128		def chat(self, message: str) -> str:
129		"""
130		Process a chat message, showing the execution steps transparently.
131
132		Args:
133		message: User message
134
135		Returns:
136		str: Response message
137		"""
138		# Create a strong system prompt that directs the model's behavior
139		system_prompt = f"""
140		You are an AI assistant with transparent access to various tools.
141		You always explain what you're doing step by step and show your work.
142
143		When a user asks you a question, you should:
144		1. Identify the most appropriate tool for the task
145		2. Describe what you're going to do before you do it
146		3. Show the results of any commands or functions you execute
147		4. Explain how you interpret these results
148
149		MOST IMPORTANTLY: When calculating math problems, ALWAYS use the 'calc' command rather than
150		doing the calculation yourself or using python -c. For example: calc 2+2
151
152		For searching the web:
153		- Use the 'opensearch_query' function to perform web searches
154		- Use the 'opensearch_fetch_url' function to fetch web page content safely
155		- Specify search engines if needed (available: {', '.join(self.opensearch_config['engines'])})
156
157		For file and system operations:
158		- Use 'find_files' to locate files
159		- Use 'read_file' to view file content
160		- Use 'execute_command' for running shell commands like 'ls', 'pwd', etc.
161
162		Always check command permissions carefully and explain what each command does.
163		For directory structure, use 'find .' or 'ls -la' commands.
164
165		When a user asks for information that might require searching the web or retrieving external data,
166		automatically determine if you should use the 'opensearch_query' function instead of requiring explicit search commands.
167
168		For example, if a user asks "can you give me a recipe for chilli sauce?", you should:
169		1. Recognize this as a request that requires external information
170		2. Use the opensearch_query function with an appropriate search query
171		3. Present the results in a helpful format
172
173		Only use tools when necessary - for factual questions you can answer directly, don't use search.
174		"""
175
176		try:
177		# Check if the message likely requires a search but doesn't explicitly ask for one
178		if not message.lower().startswith('search '):
179		search_indicators = [
180		"what is", "how to", "who is", "where can", "when did", "why does",
181		"recipe for", "information about", "tell me about", "find", "look up",
182		"latest", "news on", "current", "recent"
183		]
184
185		# Check if the message contains search indicators
186		needs_search = any(indicator in message.lower() for indicator in search_indicators)
187
188		# Check for question structure
189		question_structure = message.lower().startswith(("what", "how", "who", "where", "when", "why", "can", "could"))
190
191		if needs_search or question_structure:
192		# Check if message is asking for external information we likely don't have
193		factual_intent = self._detect_factual_intent(message)
194		if factual_intent:
195		self._print_info("This appears to be a question that might benefit from search. Searching automatically...")
196		try:
197		results = self.execute_search(message)
198		# Process search results before continuing with normal chat flow
199		context = self._extract_search_context(results)
200		# Add context to the user message
201		message = f"{message}\n\nSearch context: {context}"
202		except Exception as e:
203		self._print_error(f"Automatic search failed: {str(e)}")
204		self._print_info("Continuing without search results")
205
206		# Print what we're doing
207		self._print_status(f"Processing message: {message}")
208
209		# Start timer for perf measurement
210		start_time = time.time()
211
212		# Send message to the agent
213		response = self.agent.chat(
214		message=message,
215		system_prompt=system_prompt,
216		context=self.context
217		)
218
219		# Update context
220		self.context.append({"role": "user", "content": message})
221		self.context.append({"role": "assistant", "content": response})
222
223		# Limit context size
224		if len(self.context) > 10:
225		self.context = self.context[-10:]
226
227		# Calculate execution time
228		execution_time = time.time() - start_time
229		self._print_success(f"Response generated in {execution_time:.2f} seconds")
230
231		return response
232
233		except OllamaToolsError as e:
234		error_details = handle_exception(e, log_error=True)
235
236		self._print_error(f"Error: {error_details['error_message']}")
237		self._suggest_fix(self._generate_error_fix_suggestion(e))
238
239		return f"I encountered an error: {error_details['error_message']}\n\n" + \
240		f"Suggested fix: {self._generate_error_fix_suggestion(e)}"
241
242		except Exception as e:
243		error_details = handle_exception(e, log_error=True)
244
245		self._print_error(f"Unexpected error: {str(e)}")
246
247		return f"I encountered an unexpected error: {str(e)}"
248
249		def execute_search(self, query: str) -> Dict[str, Any]:
250		"""
251		Demonstrate direct search execution using OpenSearchNG.
252
253		Args:
254		query: Search query
255
256		Returns:
257		dict: Search results
258		"""
259		try:
260		self._print_status(f"Searching for: {query}")
261
262		# Execute the search
263		results = opensearch_query(
264		query=query,
265		config=self.opensearch_config,
266		permission_manager=self.agent.permission_manager
267		)
268
269		# Show the number of results
270		num_results = len(results.get("results", []))
271		self._print_success(f"Found {num_results} results")
272
273		# Show suggested answer if available
274		if results.get("answer"):
275		self._print_info(f"Direct answer: {results['answer']}")
276
277		# Register in execution history
278		self.execution_history.append({
279		"function": "opensearch_query",
280		"args": {"query": query},
281		"timestamp": time.time(),
282		"success": True,
283		"num_results": num_results
284		})
285
286		return results
287
288		except Exception as e:
289		self._print_error(f"Search failed: {str(e)}")
290
291		# Register in execution history
292		self.execution_history.append({
293		"function": "opensearch_query",
294		"args": {"query": query},
295		"timestamp": time.time(),
296		"success": False,
297		"error": str(e)
298		})
299
300		self._suggest_fix(self._generate_error_fix_suggestion(e))
301		raise
302
303		def execute_command(self, command: str, args: Optional[List[str]] = None) -> Dict[str, Any]:
304		"""
305		Execute a system command with transparency.
306
307		Args:
308		command: Command to execute
309		args: Command arguments
310
311		Returns:
312		dict: Command result
313		"""
314		try:
315		# Show what we're doing
316		cmd_str = f"{command} {' '.join(args) if args else ''}"
317		self._print_status(f"Executing command: {cmd_str}")
318
319		# Execute the command
320		from ollamatools.tools.system_tools import execute_command
321		result = execute_command(
322		command=command,
323		args=args,
324		permission_manager=self.agent.permission_manager,
325		custom_allowlist=self.command_allowlist
326		)
327
328		# Check for errors
329		if "error" in result:
330		self._print_error(f"Command error: {result['error']}")
331
332		# Register in execution history
333		self.execution_history.append({
334		"function": "execute_command",
335		"args": {"command": command, "args": args},
336		"timestamp": time.time(),
337		"success": False,
338		"error": result["error"]
339		})
340
341		self._suggest_fix(self._generate_command_fix_suggestion(result))
342		return result
343
344		# Show success
345		self._print_success(f"Command executed with exit code: {result['returncode']}")
346
347		# Format the output for better readability
348		if result.get('stdout'):
349		output_lines = result['stdout'].strip().split('\n')
350		if len(output_lines) > 10:
351		# Truncate long output
352		preview = '\n'.join(output_lines[:10])
353		self._print_info(f"Command output (first 10 lines):\n{preview}\n[...more output truncated...]")
354		else:
355		self._print_info(f"Command output:\n{result['stdout'].strip()}")
356
357		# Show errors if any
358		if result.get('stderr'):
359		self._print_warning(f"Command stderr:\n{result['stderr'].strip()}")
360
361		# Register in execution history
362		self.execution_history.append({
363		"function": "execute_command",
364		"args": {"command": command, "args": args},
365		"timestamp": time.time(),
366		"success": True,
367		"returncode": result["returncode"]
368		})
369
370		return result
371
372		except Exception as e:
373		self._print_error(f"Command execution failed: {str(e)}")
374
375		# Register in execution history
376		self.execution_history.append({
377		"function": "execute_command",
378		"args": {"command": command, "args": args},
379		"timestamp": time.time(),
380		"success": False,
381		"error": str(e)
382		})
383
384		self._suggest_fix(self._generate_error_fix_suggestion(e))
385		raise
386
387		def get_directory_structure(self, path: str = '.', max_depth: int = 2) -> Dict[str, Any]:
388		"""
389		Get a formatted directory structure.
390
391		Args:
392		path: Starting directory path
393		max_depth: Maximum directory depth to show
394
395		Returns:
396		dict: Directory structure information
397		"""
398		try:
399		self._print_status(f"Getting directory structure for: {path} (depth: {max_depth})")
400
401		# Execute find command to get the structure
402		result = self.execute_command(
403		command="find",
404		args=[path, "-maxdepth", str(max_depth), "-type", "f", "-o", "-type", "d"]
405		)
406
407		if "error" in result:
408		return result
409
410		# Process the output into a more structured format
411		lines = result.get("stdout", "").strip().split('\n')
412
413		structure = {
414		"path": path,
415		"depth": max_depth,
416		"entries": lines,
417		"files": [line for line in lines if os.path.isfile(line)],
418		"directories": [line for line in lines if os.path.isdir(line)],
419		"count": {
420		"total": len(lines),
421		"files": len([line for line in lines if os.path.isfile(line)]),
422		"directories": len([line for line in lines if os.path.isdir(line)])
423		}
424		}
425
426		self._print_success(f"Found {structure['count']['total']} entries "
427		f"({structure['count']['files']} files, "
428		f"{structure['count']['directories']} directories)")
429
430		return structure
431
432		except Exception as e:
433		self._print_error(f"Failed to get directory structure: {str(e)}")
434		self._suggest_fix(self._generate_error_fix_suggestion(e))
435
436		return {
437		"error": str(e),
438		"path": path,
439		"depth": max_depth
440		}
441
442		def show_execution_history(self, max_entries: int = 10) -> None:
443		"""
444		Display the execution history.
445
446		Args:
447		max_entries: Maximum number of history entries to show
448		"""
449		if not self.execution_history:
450		print("\n📜 Execution history is empty.")
451		return
452
453		# Get the most recent entries
454		recent_entries = self.execution_history[-max_entries:]
455
456		print("\n📜 Execution History:")
457		print("=" * 60)
458
459		for i, entry in enumerate(recent_entries, 1):
460		# Format timestamp
461		timestamp = time.strftime("%H:%M:%S", time.localtime(entry["timestamp"]))
462
463		# Format function and args
464		function_name = entry["function"]
465		args_str = ", ".join(f"{k}={v!r}" for k, v in entry["args"].items())
466
467		# Format status
468		status = "✅ Success" if entry.get("success", False) else "❌ Failed"
469
470		# Print the entry
471		print(f"{i}. [{timestamp}] {function_name}({args_str}) -> {status}")
472
473		# Show additional details for errors
474		if not entry.get("success", False) and "error" in entry:
475		print(f" Error: {entry['error']}")
476
477		# Show additional details for commands
478		if function_name == "execute_command" and "returncode" in entry:
479		print(f" Exit code: {entry['returncode']}")
480
481		# Show additional details for search
482		if function_name == "opensearch_query" and "num_results" in entry:
483		print(f" Results: {entry['num_results']}")
484
485		print("=" * 60)
486
487		def _generate_error_fix_suggestion(self, error: Exception) -> str:
488		"""
489		Generate a fix suggestion based on the error type.
490
491		Args:
492		error: The exception
493
494		Returns:
495		str: Suggested fix
496		"""
497		if isinstance(error, CommandError):
498		return self._generate_command_fix_suggestion(error)
499		elif isinstance(error, FileError):
500		return self._generate_file_fix_suggestion(error)
501		elif isinstance(error, WebError):
502		return self._generate_web_fix_suggestion(error)
503		else:
504		# Generic suggestion
505		return f"Try a different approach or check if the required resources are available."
506
507		def _generate_command_fix_suggestion(self, error: Union[CommandError, Dict[str, Any]]) -> str:
508		"""
509		Generate a fix suggestion for command errors.
510
511		Args:
512		error: Error object or dictionary
513
514		Returns:
515		str: Suggested fix
516		"""
517		# Extract error details
518		if isinstance(error, CommandError):
519		error_msg = str(error)
520		details = error.details or {}
521		command = details.get('command', '')
522		stderr = details.get('stderr', '')
523		else:
524		error_msg = error.get('error', '')
525		command = error.get('command', '')
526		stderr = error.get('stderr', '')
527
528		# Check for common error patterns
529		if "not in allowed command list" in error_msg:
530		return (f"The command '{command}' is not in the allowed command list. "
531		f"Try using one of these commands instead: {', '.join(sorted(list(self.command_allowlist)[:10]))}... "
532		f"or add this command to the allowlist when initializing the agent.")
533
534		elif "dangerous pattern" in error_msg:
535		pattern = error_msg.split("pattern matching '")[1].split("'")[0] if "pattern matching '" in error_msg else ""
536		return (f"The command contains potentially dangerous pattern: {pattern}. "
537		f"Try reformulating the command to avoid shell metacharacters.")
538
539		elif "not found" in error_msg or "command not found" in stderr:
540		return f"The command '{command}' does not exist on this system. Check the spelling or install the command."
541
542		elif "permission denied" in stderr.lower():
543		return f"Permission denied when executing '{command}'. Try running with appropriate permissions."
544
545		else:
546		return f"Try a different command syntax or check command documentation."
547
548		def _generate_file_fix_suggestion(self, error: FileError) -> str:
549		"""
550		Generate a fix suggestion for file errors.
551
552		Args:
553		error: File error
554
555		Returns:
556		str: Suggested fix
557		"""
558		error_msg = str(error)
559		details = error.details or {}
560		path = details.get('path', '')
561
562		if "not found" in error_msg:
563		return f"The file '{path}' does not exist. Check the path or create the file."
564
565		elif "is a directory" in error_msg:
566		return f"'{path}' is a directory, not a file. Try specifying a file path instead."
567
568		elif "permission denied" in error_msg.lower():
569		return f"Permission denied when accessing '{path}'. Check file permissions."
570
571		elif "exceeds maximum size" in error_msg:
572		return f"The file '{path}' is too large. Try reading a smaller file or increasing max_size."
573
574		elif "outside the allowed directories" in error_msg:
575		return f"The path '{path}' is outside the allowed directories. Try a path within the sandbox."
576
577		else:
578		return f"Check if the file exists and you have appropriate permissions."
579
580		def _generate_web_fix_suggestion(self, error: WebError) -> str:
581		"""
582		Generate a fix suggestion for web errors.
583
584		Args:
585		error: Web error
586
587		Returns:
588		str: Suggested fix
589		"""
590		error_msg = str(error)
591		details = error.details or {}
592		url = details.get('url', '')
593
594		if "domain" in error_msg and "blocked" in error_msg:
595		return f"The domain in '{url}' is blocked for security reasons. Try a different domain."
596
597		elif "scheme" in error_msg and "not allowed" in error_msg:
598		return f"The URL scheme is not allowed. Use http:// or https:// instead."
599
600		elif "Connection refused" in error_msg:
601		return f"Could not connect to '{url}'. Check if the service is running and accessible."
602
603		elif "timeout" in error_msg.lower():
604		return f"The request to '{url}' timed out. Try again later or check your internet connection."
605
606		elif "Status code:" in error_msg and "404" in error_msg:
607		return f"The resource at '{url}' was not found (404). Check the URL for typos."
608
609		elif "Status code:" in error_msg and "403" in error_msg:
610		return f"Access to '{url}' is forbidden (403). The site may require authentication."
611
612		else:
613		return f"Try a different URL or check your internet connection."
614
615		def _print_status(self, message: str) -> None:
616		"""Print a status message."""
617		print(f"\n🔄 {message}")
618
619		def _print_success(self, message: str) -> None:
620		"""Print a success message."""
621		print(f"✅ {message}")
622
623		def _print_error(self, message: str) -> None:
624		"""Print an error message."""
625		print(f"❌ {message}")
626
627		def _print_warning(self, message: str) -> None:
628		"""Print a warning message."""
629		print(f"⚠️ {message}")
630
631		def _print_info(self, message: str) -> None:
632		"""Print an information message."""
633		print(f"ℹ️ {message}")
634
635		def _suggest_fix(self, suggestion: str) -> None:
636		"""Print a fix suggestion."""
637		print(f"💡 Suggested fix: {suggestion}")
638
639		def _detect_factual_intent(self, message: str) -> bool:
640		"""
641		Detect if a message is likely asking for factual information
642		that would benefit from search.
643
644		Args:
645		message: User message
646
647		Returns:
648		bool: True if message likely needs external information
649		"""
650		# Basic topics that often require external information
651		external_topics = [
652		"recipe", "weather", "news", "sports", "movie", "book", "author",
653		"product", "review", "price", "stock", "company", "history",
654		"person", "celebrity", "politician", "event", "scientific",
655		"research", "study", "statistic", "data", "definition"
656		]
657
658		# Check if message contains any of these topics
659		contains_topic = any(topic in message.lower() for topic in external_topics)
660
661		# Check if it's a complex question (longer questions tend to be more specific)
662		is_complex = len(message.split()) > 6
663
664		# Simple heuristic: if it contains a topic of interest and is reasonably complex
665		return contains_topic and is_complex
666
667		def _extract_search_context(self, search_results: Dict[str, Any]) -> str:
668		"""
669		Extract useful context from search results.
670
671		Args:
672		search_results: Results from opensearch_query
673
674		Returns:
675		str: Formatted context for the agent
676		"""
677		context_parts = []
678
679		# Add direct answer if available
680		if search_results.get("answer"):
681		context_parts.append(f"Direct answer: {search_results['answer']}")
682
683		# Add snippets from results
684		if search_results.get("results"):
685		context_parts.append("Information from search results:")
686		for i, result in enumerate(search_results["results"][:3]): # Limit to 3 results
687		title = result.get("title", "Untitled")
688		content = result.get("content", "No content")
689		url = result.get("url", "")
690		context_parts.append(f"- {title}: {content} (Source: {url})")
691
692		# Join everything with newlines
693		return "\n".join(context_parts)
694
695		def list_allowed_commands(self) -> List[str]:
696		"""
697		List all currently allowed commands.
698
699		Returns:
700		List[str]: List of allowed commands
701		"""
702		# Get the current allowlist from command_validator
703		from ollamatools.command_validator import DEFAULT_ALLOWLIST
704
705		# Combine with any custom entries
706		all_commands = sorted(list(self.command_allowlist))
707
708		# Print the commands
709		self._print_info(f"Currently allowed commands ({len(all_commands)}):")
710
711		# Format into categories for better readability
712		categories = {
713		"File Operations": ["ls", "cat", "find", "grep", "head", "tail"],
714		"Directory Operations": ["pwd", "cd", "mkdir", "rmdir"],
715		"System Information": ["uname", "hostname", "whoami", "ps", "top", "df", "du", "free"],
716		"Network Utilities": ["ping", "curl", "wget", "netstat", "dig", "nslookup"],
717		"Text Processing": ["echo", "printf", "sed", "awk", "cut", "sort", "uniq", "wc"],
718		"File Manipulation": ["cp", "mv", "rm", "touch", "chmod", "chown", "chgrp", "ln"],
719		"Other": []
720		}
721
722		# Categorize commands
723		for cmd in all_commands:
724		categorized = False
725		for category, commands in categories.items():
726		if cmd in commands:
727		categorized = True
728		break
729		if not categorized:
730		categories["Other"].append(cmd)
731
732		# Print by category
733		for category, commands in categories.items():
734		if commands:
735		print(f"\n{category}:")
736		for i, cmd in enumerate(sorted(commands)):
737		print(f" {cmd}", end="\t")
738		# New line every 5 commands for readability
739		if (i + 1) % 5 == 0:
740		print()
741		print() # Ensure we end with a newline
742
743		return all_commands
744
745		def add_allowed_command(self, command: str) -> bool:
746		"""
747		Add a command to the allowed list.
748
749		Args:
750		command: Command to add
751
752		Returns:
753		bool: True if added successfully
754		"""
755		if not command or not isinstance(command, str):
756		self._print_error("Invalid command name")
757		return False
758
759		# Clean up the command name
760		command = command.strip().lower()
761
762		# Check if already in allowlist
763		if command in self.command_allowlist:
764		self._print_info(f"Command '{command}' is already allowed")
765		return True
766
767		# Add to custom allowlist
768		self.command_allowlist.add(command)
769		self._print_success(f"Command '{command}' added to allowed list")
770
771		# Show warning for potentially dangerous commands
772		dangerous_commands = {"rm", "mv", "chmod", "chown", "dd", "mkfs", "iptables"}
773		if command in dangerous_commands:
774		self._print_warning(f"'{command}' is a potentially dangerous command. Use with caution!")
775
776		return True
777
778		def remove_allowed_command(self, command: str) -> bool:
779		"""
780		Remove a command from the allowed list.
781
782		Args:
783		command: Command to remove
784
785		Returns:
786		bool: True if removed successfully
787		"""
788		if not command or not isinstance(command, str):
789		self._print_error("Invalid command name")
790		return False
791
792		# Clean up the command name
793		command = command.strip().lower()
794
795		# Check if in allowlist
796		if command not in self.command_allowlist:
797		self._print_error(f"Command '{command}' is not in the allowed list")
798		return False
799
800		# Remove from custom allowlist
801		self.command_allowlist.remove(command)
802		self._print_success(f"Command '{command}' removed from allowed list")
803
804		return True
805
806		def _load_issues(self) -> None:
807		"""Load saved issues from file."""
808		if os.path.exists(self.issues_file):
809		try:
810		with open(self.issues_file, 'r') as f:
811		data = json.load(f)
812		self.issues = data.get('issues', [])
813		self.issue_id_counter = data.get('next_id', 1)
814		logger.info(f"Loaded {len(self.issues)} issues from {self.issues_file}")
815		except Exception as e:
816		logger.error(f"Failed to load issues: {str(e)}")
817		self.issues = []
818		self.issue_id_counter = 1
819
820		def _save_issues(self) -> None:
821		"""Save issues to file."""
822		try:
823		with open(self.issues_file, 'w') as f:
824		json.dump({
825		'issues': self.issues,
826		'next_id': self.issue_id_counter
827		}, f, indent=2)
828		logger.info(f"Saved {len(self.issues)} issues to {self.issues_file}")
829		except Exception as e:
830		logger.error(f"Failed to save issues: {str(e)}")
831
832		def add_issue(self, title: str, description: str, severity: str = 'medium') -> int:
833		"""
834		Add a new issue to the tracker.
835
836		Args:
837		title: Issue title
838		description: Issue description
839		severity: Issue severity (low, medium, high)
840
841		Returns:
842		int: Issue ID
843		"""
844		# Validate severity
845		if severity not in ['low', 'medium', 'high']:
846		severity = 'medium'
847
848		# Create issue
849		issue = {
850		'id': self.issue_id_counter,
851		'title': title,
852		'description': description,
853		'severity': severity,
854		'status': 'open',
855		'created_at': time.strftime('%Y-%m-%d %H:%M:%S'),
856		'updated_at': time.strftime('%Y-%m-%d %H:%M:%S')
857		}
858
859		# Add to list and increment counter
860		self.issues.append(issue)
861		self.issue_id_counter += 1
862
863		# Save issues
864		self._save_issues()
865
866		self._print_success(f"Issue #{issue['id']} added: {title}")
867		return issue['id']
868
869		def list_issues(self, status: Optional[str] = None) -> List[Dict[str, Any]]:
870		"""
871		List all issues, optionally filtered by status.
872
873		Args:
874		status: Optional status filter (open, resolved)
875
876		Returns:
877		List[Dict[str, Any]]: List of issues
878		"""
879		# Filter issues if status provided
880		filtered_issues = self.issues
881		if status:
882		filtered_issues = [i for i in self.issues if i['status'] == status]
883
884		# Print issues
885		if not filtered_issues:
886		self._print_info("No issues found")
887		return []
888
889		self._print_info(f"Issues ({len(filtered_issues)}):")
890		print("\nID \| Severity \| Status \| Title")
891		print("----\|----------\|----------\|-------------------")
892
893		for issue in filtered_issues:
894		print(f"{issue['id']:<4}\| {issue['severity']:<9}\| {issue['status']:<9}\| {issue['title']}")
895
896		return filtered_issues
897
898		def get_issue(self, issue_id: int) -> Optional[Dict[str, Any]]:
899		"""
900		Get details of a specific issue.
901
902		Args:
903		issue_id: Issue ID
904
905		Returns:
906		Optional[Dict[str, Any]]: Issue details or None if not found
907		"""
908		# Find issue
909		for issue in self.issues:
910		if issue['id'] == issue_id:
911		# Print issue details
912		self._print_info(f"Issue #{issue['id']}: {issue['title']}")
913		print(f"Status: {issue['status']}")
914		print(f"Severity: {issue['severity']}")
915		print(f"Created: {issue['created_at']}")
916		print(f"Updated: {issue['updated_at']}")
917		print("\nDescription:")
918		print(issue['description'])
919		return issue
920
921		self._print_error(f"Issue #{issue_id} not found")
922		return None
923
924		def update_issue(self, issue_id: int, status: Optional[str] = None,
925		severity: Optional[str] = None, description: Optional[str] = None) -> bool:
926		"""
927		Update an existing issue.
928
929		Args:
930		issue_id: Issue ID
931		status: New status (open, resolved)
932		severity: New severity (low, medium, high)
933		description: New description
934
935		Returns:
936		bool: True if updated successfully
937		"""
938		# Find issue
939		for i, issue in enumerate(self.issues):
940		if issue['id'] == issue_id:
941		# Update fields
942		if status and status in ['open', 'resolved']:
943		self.issues[i]['status'] = status
944
945		if severity and severity in ['low', 'medium', 'high']:
946		self.issues[i]['severity'] = severity
947
948		if description:
949		self.issues[i]['description'] = description
950
951		# Update timestamp
952		self.issues[i]['updated_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
953
954		# Save issues
955		self._save_issues()
956
957		self._print_success(f"Issue #{issue_id} updated")
958		return True
959
960		self._print_error(f"Issue #{issue_id} not found")
961		return False
962
963
964		def main():
965		"""Run the transparent Ollama agent."""
966		try:
967		# Initialize the agent
968		logger.info("Initializing TransparentOllamaAgent...")
969		agent = TransparentOllamaAgent(
970		model="llama3.1:latest", # Adjust according to your available models
971		permission_prompts=True
972		)
973
974		logger.info("Agent initialized. Enter 'exit' to quit.")
975		print("\nTransparent Ollama Agent")
976		print("========================")
977		print("Type 'exit' to quit")
978		print("Type 'commands' to see available commands")
979		print("Type 'add_command <name>' to add a command to the allowed list")
980		print("Type 'remove_command <name>' to remove a command from the allowed list")
981		print("Type 'search <query>' to search directly")
982		print("Type 'run <command>' to execute a command directly")
983		print("Type 'dir' to show current directory structure")
984		print("Type 'history' to see execution history")
985
986		print("\nIssue Tracking:")
987		print(" issues - List all issues")
988		print(" issue <id> - Show issue details")
989		print(" add_issue <title> \| <description> \| [severity] - Add a new issue")
990		print(" resolve_issue <id> - Mark an issue as resolved")
991
992		print("\nOtherwise, just chat naturally with the agent!")
993
994		# Simple chat loop
995		while True:
996		# Get user input
997		user_input = input("\nYou: ").strip()
998
999		# Check for exit command
1000		if user_input.lower() in ['exit', 'quit', 'bye']:
1001		print("\nExiting. Goodbye!")
1002		break
1003
1004		# Check for special commands
1005		if user_input.lower() == 'commands':
1006		agent.list_allowed_commands()
1007		continue
1008
1009		if user_input.lower().startswith('add_command '):
1010		command = user_input[len('add_command '):].strip()
1011		agent.add_allowed_command(command)
1012		continue
1013
1014		if user_input.lower().startswith('remove_command '):
1015		command = user_input[len('remove_command '):].strip()
1016		agent.remove_allowed_command(command)
1017		continue
1018
1019		if user_input.lower() == 'history':
1020		agent.show_execution_history()
1021		continue
1022
1023		if user_input.lower().startswith('search '):
1024		query = user_input[7:].strip()
1025		if query:
1026		try:
1027		results = agent.execute_search(query)
1028		# Show the top 3 results
1029		if results.get("results"):
1030		print("\nTop Results:")
1031		for i, result in enumerate(results["results"][:3], 1):
1032		print(f"{i}. {result['title']}")
1033		print(f" URL: {result['url']}")
1034		print(f" {result['content'][:100]}...")
1035		print()
1036		except Exception as e:
1037		print(f"Search failed: {str(e)}")
1038		else:
1039		print("Please provide a search query.")
1040		continue
1041
1042		if user_input.lower().startswith('run '):
1043		cmd = user_input[4:].strip()
1044		if cmd:
1045		try:
1046		# Split the command into parts
1047		parts = cmd.split()
1048		if parts:
1049		main_cmd = parts[0]
1050		args = parts[1:] if len(parts) > 1 else []
1051		result = agent.execute_command(main_cmd, args)
1052		if "stdout" in result:
1053		print("\nCommand Output:")
1054		print(result["stdout"])
1055		else:
1056		print("Please provide a command to run.")
1057		except Exception as e:
1058		print(f"Command execution failed: {str(e)}")
1059		continue
1060
1061		if user_input.lower().startswith('dir'):
1062		parts = user_input.split()
1063		path = parts[1] if len(parts) > 1 else '.'
1064		depth = int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else 2
1065
1066		try:
1067		agent.get_directory_structure(path, depth)
1068		except Exception as e:
1069		print(f"Failed to get directory structure: {str(e)}")
1070		continue
1071
1072		# Issue tracking commands
1073		if user_input.lower() == 'issues':
1074		agent.list_issues()
1075		continue
1076
1077		if user_input.lower().startswith('issue '):
1078		try:
1079		issue_id = int(user_input[len('issue '):].strip())
1080		agent.get_issue(issue_id)
1081		except ValueError:
1082		print("Invalid issue ID. Use 'issue <number>'.")
1083		continue
1084
1085		if user_input.lower().startswith('add_issue '):
1086		# Parse command: add_issue Title \| Description \| [severity]
1087		parts = user_input[len('add_issue '):].split('\|')
1088		if len(parts) >= 2:
1089		title = parts[0].strip()
1090		description = parts[1].strip()
1091		severity = parts[2].strip().lower() if len(parts) > 2 else 'medium'
1092		agent.add_issue(title, description, severity)
1093		else:
1094		print("Usage: add_issue Title \| Description \| [severity]")
1095		continue
1096
1097		if user_input.lower().startswith('resolve_issue '):
1098		try:
1099		issue_id = int(user_input[len('resolve_issue '):].strip())
1100		agent.update_issue(issue_id, status='resolved')
1101		except ValueError:
1102		print("Invalid issue ID. Use 'resolve_issue <number>'.")
1103		continue
1104
1105		# Process as normal chat message
1106		response = agent.chat(user_input)
1107		print(f"\nAssistant: {response}")
1108
1109		except KeyboardInterrupt:
1110		print("\n\nExiting due to keyboard interrupt.")
1111		except Exception as e:
1112		logger.error(f"Unexpected error in main function: {str(e)}")
1113		print(f"\nUnexpected error: {str(e)}")
1114		print("Check the logs for details.")
1115
1116		return 0
1117
1118		if __name__ == "__main__":
1119		sys.exit(main())

ollamatools-project/ollamatools/agent/__init__.py +1 -5

…	…	@@ -10,11 +10,7 @@ from .modular import ModularOllamaAgent
10	10	# Import the ExtendedModularAgent
11	11	from .extended import ExtendedModularAgent
12	12
13		# For backward compatibility
14		from .transparent import TransparentOllamaAgent
15
16	13	__all__ = [
17	14	'ModularOllamaAgent',
18		'ExtendedModularAgent',
19		'TransparentOllamaAgent'
	15	'ExtendedModularAgent'
20	16	]

ollamatools-project/ollamatools/agent/transparent/__init__.py +0 -74

…	…	@@ -1,74 +0,0 @@
1		"""
2		OllamaTools Transparent Agent module initialization.
3
4		This package provides an enhanced OllamaAgent that offers transparent execution,
5		better feedback, and additional features like search integration, command
6		management, issue tracking, and semantic function selection feedback.
7
8		Note: This module is maintained for backward compatibility. New code should
9		use the ModularOllamaAgent from ollamatools.agent.modular instead.
10		"""
11
12		import warnings
13		import sys
14		from ..modular import (
15		ModularOllamaAgent,
16		print_status, print_success, print_error, print_warning, print_info, suggest_fix,
17		FunctionSelectionFeedback,
18		detect_factual_intent, extract_search_context, execute_search,
19		list_allowed_commands, add_allowed_command, remove_allowed_command,
20		load_issues, save_issues, add_issue, list_issues, get_issue, update_issue
21		)
22		from ..modular.feedback import (
23		detect_command_intent,
24		extract_command_from_similar_queries,
25		format_command_result
26		)
27
28		# For backward compatibility, alias ModularOllamaAgent as TransparentOllamaAgent
29		TransparentOllamaAgent = ModularOllamaAgent
30
31		# Show deprecation warning when TransparentOllamaAgent is imported
32		warnings.warn(
33		"TransparentOllamaAgent is deprecated and will be removed in a future version. "
34		"Use ModularOllamaAgent from ollamatools.agent.modular instead.",
35		DeprecationWarning,
36		stacklevel=2
37		)
38
39		# For backward compatibility, also import from the original modules
40		# This is needed for code that directly imports from these modules
41		try:
42		from .agent import TransparentOllamaAgent as _OriginalTransparentOllamaAgent
43		from .search import detect_factual_intent as _original_detect_factual_intent
44		from .commands import list_allowed_commands as _original_list_allowed_commands
45		# ... and so on for other imports
46		except ImportError:
47		# If the original modules don't exist, that's fine
48		pass
49
50		__all__ = [
51		'TransparentOllamaAgent',
52		'detect_factual_intent',
53		'extract_search_context',
54		'execute_search',
55		'list_allowed_commands',
56		'add_allowed_command',
57		'remove_allowed_command',
58		'load_issues',
59		'save_issues',
60		'add_issue',
61		'list_issues',
62		'get_issue',
63		'update_issue',
64		'print_status',
65		'print_success',
66		'print_error',
67		'print_warning',
68		'print_info',
69		'suggest_fix',
70		'FunctionSelectionFeedback',
71		'detect_command_intent',
72		'extract_command_from_similar_queries',
73		'format_command_result'
74		]

ollamatools-project/ollamatools/agent/transparent/agent.py +0 -697

…	…	@@ -1,697 +0,0 @@
1		"""
2		Transparent OllamaAgent implementation.
3
4		This module provides the TransparentOllamaAgent class, which extends
5		the base OllamaAgent with transparent execution, better feedback, and
6		additional features like search integration, command management,
7		issue tracking, and semantic function selection feedback.
8		"""
9
10		import os
11		import sys
12		import json
13		import time
14		import logging
15		import re
16		from typing import Dict, Any, List, Optional, Set, Union, Tuple
17
18		from ollamatools import OllamaAgent, Config
19		from ollamatools.command_validator import DEFAULT_ALLOWLIST
20		from ollamatools.utils.logger import setup_logger
21		from ollamatools.opensearch_web_tools import (
22		opensearch_query,
23		opensearch_fetch_url,
24		get_opensearch_info,
25		DEFAULT_OPENSEARCH_CONFIG
26		)
27		from ollamatools.errors import (
28		OllamaToolsError,
29		CommandError,
30		FileError,
31		WebError,
32		handle_exception
33		)
34		from ollamatools.tools.system_tools import execute_command
35
36		# Import agent components
37		from .search import detect_factual_intent, extract_search_context, execute_search
38		from .commands import list_allowed_commands, add_allowed_command, remove_allowed_command
39		from .issues import load_issues, save_issues, add_issue, list_issues, get_issue, update_issue
40		from .utils import print_status, print_success, print_error, print_warning, print_info, suggest_fix
41		from .feedback import (
42		FunctionSelectionFeedback,
43		detect_command_intent,
44		extract_command_from_similar_queries,
45		format_command_result
46		)
47
48		# Set up logging
49		logger = setup_logger("transparent_ollama_agent", level="INFO")
50
51		# Extend the default command allowlist with useful commands
52		EXTENDED_ALLOWLIST = DEFAULT_ALLOWLIST.union({
53		"python", "python3", "calc", "bc", "find", "grep", "awk", "sed",
54		"head", "tail", "more", "less", "cat", "echo", "cd", "pwd"
55		})
56
57		# Configuration for OpenSearchNG
58		OPENSEARCH_CONFIG = DEFAULT_OPENSEARCH_CONFIG.copy()
59		OPENSEARCH_CONFIG.update({
60		"base_url": "http://localhost:8081", # Updated to your container's port
61		"engines": ["google", "bing", "duckduckgo", "brave"],
62		"num_results": 5
63		})
64
65		class TransparentOllamaAgent:
66		"""
67		An enhanced OllamaAgent that provides transparent execution and better feedback.
68		"""
69
70		def __init__(
71		self,
72		model: str = "llama3.1:latest",
73		permission_prompts: bool = True,
74		opensearch_config: Optional[Dict[str, Any]] = None,
75		command_allowlist: Optional[Set[str]] = None,
76		feedback_file_path: Optional[str] = None
77		):
78		"""
79		Initialize the agent.
80
81		Args:
82		model: Ollama model to use
83		permission_prompts: Whether to display permission prompts
84		opensearch_config: OpenSearchNG configuration
85		command_allowlist: Custom command allowlist
86		feedback_file_path: Path to the feedback file (default: ~/.ollamatools/function_selection_feedback.jsonl)
87		"""
88		# Initialize OllamaAgent
89		self.agent = OllamaAgent(
90		model=model,
91		permission_prompts=permission_prompts
92		)
93
94		# Store configuration
95		self.opensearch_config = opensearch_config or OPENSEARCH_CONFIG
96		self.command_allowlist = command_allowlist or EXTENDED_ALLOWLIST
97
98		# Initialize state
99		self.context = []
100		self.execution_history = []
101
102		# Initialize issue tracking
103		self.issues = []
104		self.issue_id_counter = 1
105
106		# Path for saving issues
107		self.issues_file = os.path.join(
108		os.path.expanduser('~'),
109		'.ollamatools',
110		'issues.json'
111		)
112
113		# Create directory if it doesn't exist
114		os.makedirs(os.path.dirname(self.issues_file), exist_ok=True)
115
116		# Load existing issues if any
117		self.load_issues()
118
119		# Initialize function selection feedback system
120		self.function_feedback = FunctionSelectionFeedback(feedback_file_path)
121		logger.info("Initialized function selection feedback system")
122
123		# Define system prompt
124		self.system_prompt = f"""
125		You are an AI assistant with transparent access to various tools.
126		You always explain what you're doing step by step and show your work.
127
128		When a user asks you a question, you should:
129		1. Identify the most appropriate tool for the task
130		2. Describe what you're going to do before you do it
131		3. Show the results of any commands or functions you execute
132		4. Explain how you interpret these results
133
134		MOST IMPORTANTLY: When calculating math problems, ALWAYS use the 'calc' command rather than
135		doing the calculation yourself or using python -c. For example: calc 2+2
136
137		For searching the web:
138		- Use the 'opensearch_query' function to perform web searches
139		- Use the 'opensearch_fetch_url' function to fetch web page content safely
140		- Specify search engines if needed (available: {', '.join(self.opensearch_config['engines'])})
141
142		For file and system operations:
143		- Use 'find_files' to locate files
144		- Use 'read_file' to view file content
145		- Use 'execute_command' for running shell commands like 'ls', 'pwd', etc.
146
147		For system information queries:
148		- When a user asks about the current directory, use 'pwd' command
149		- When a user asks about file listings, use 'ls' command
150		- When a user asks about the current user, use 'whoami' command
151		- When a user asks about system information, prefer shell commands over other methods
152
153		Always check command permissions carefully and explain what each command does.
154		For directory structure, use 'find .' or 'ls -la' commands.
155
156		When a user asks for information that might require searching the web or retrieving external data,
157		automatically determine if you should use the 'opensearch_query' function instead of requiring explicit search commands.
158
159		For example, if a user asks "can you give me a recipe for chilli sauce?", you should:
160		1. Recognize this as a request that requires external information
161		2. Use the opensearch_query function with an appropriate search query
162		3. Present the results in a helpful format
163
164		Only use tools when necessary - for factual questions you can answer directly, don't use search.
165		"""
166
167		# Check if OpenSearchNG is available
168		try:
169		self.print_status("Checking OpenSearchNG availability...")
170		info = get_opensearch_info(config=self.opensearch_config)
171		self.print_success(f"OpenSearchNG running at {self.opensearch_config['base_url']}")
172
173		# Log available engines
174		if 'engines' in info and info['engines']:
175		self.print_info(f"Available search engines: {', '.join(info['engines'])}")
176		except Exception as e:
177		self.print_error(f"OpenSearchNG not available: {str(e)}")
178		self.print_warning("Falling back to regular search methods")
179		self.suggest_fix("Make sure your OpenSearchNG instance is running at " +
180		f"{self.opensearch_config['base_url']}")
181
182		# Utility methods redirected to utils module
183		def print_status(self, message: str) -> None:
184		print_status(message)
185
186		def print_success(self, message: str) -> None:
187		print_success(message)
188
189		def print_error(self, message: str) -> None:
190		print_error(message)
191
192		def print_warning(self, message: str) -> None:
193		print_warning(message)
194
195		def print_info(self, message: str) -> None:
196		print_info(message)
197
198		def suggest_fix(self, suggestion: str) -> None:
199		suggest_fix(suggestion)
200
201		# Search methods redirected to search module
202		def detect_factual_intent(self, message: str) -> bool:
203		return detect_factual_intent(message)
204
205		def extract_search_context(self, search_results: Dict[str, Any]) -> str:
206		return extract_search_context(search_results)
207
208		def execute_search(self, query: str) -> Dict[str, Any]:
209		return execute_search(query, self.opensearch_config, self.agent.permission_manager, self.execution_history)
210
211		# Command management methods redirected to commands module
212		def list_allowed_commands(self) -> List[str]:
213		return list_allowed_commands(self.command_allowlist)
214
215		def add_allowed_command(self, command: str) -> bool:
216		return add_allowed_command(command, self.command_allowlist)
217
218		def remove_allowed_command(self, command: str) -> bool:
219		return remove_allowed_command(command, self.command_allowlist)
220
221		# Issue tracking methods redirected to issues module
222		def load_issues(self) -> None:
223		load_issues(self)
224
225		def save_issues(self) -> None:
226		save_issues(self)
227
228		def add_issue(self, title: str, description: str, severity: str = 'medium') -> int:
229		return add_issue(self, title, description, severity)
230
231		def list_issues(self, status: Optional[str] = None) -> List[Dict[str, Any]]:
232		return list_issues(self, status)
233
234		def get_issue(self, issue_id: int) -> Optional[Dict[str, Any]]:
235		return get_issue(self, issue_id)
236
237		def update_issue(self, issue_id: int, status: Optional[str] = None,
238		severity: Optional[str] = None, description: Optional[str] = None) -> bool:
239		return update_issue(self, issue_id, status, severity, description)
240
241		def show_execution_history(self, max_entries: int = 10) -> None:
242		"""
243		Display the execution history.
244
245		Args:
246		max_entries: Maximum number of history entries to show
247		"""
248		if not self.execution_history:
249		print("\n📜 Execution history is empty.")
250		return
251
252		# Get the most recent entries
253		recent_entries = self.execution_history[-max_entries:]
254
255		print("\n📜 Execution History:")
256		print("=" * 60)
257
258		for i, entry in enumerate(recent_entries, 1):
259		# Format timestamp
260		timestamp = time.strftime("%H:%M:%S", time.localtime(entry["timestamp"]))
261
262		# Format function and args
263		function_name = entry["function"]
264		args_str = ", ".join(f"{k}={v!r}" for k, v in entry["args"].items())
265
266		# Format status
267		status = "✅ Success" if entry.get("success", False) else "❌ Failed"
268
269		# Print the entry
270		print(f"{i}. [{timestamp}] {function_name}({args_str}) -> {status}")
271
272		# Show additional details for errors
273		if not entry.get("success", False) and "error" in entry:
274		print(f" Error: {entry['error']}")
275
276		# Show additional details for commands
277		if function_name == "execute_command" and "returncode" in entry:
278		print(f" Exit code: {entry['returncode']}")
279
280		# Show additional details for search
281		if function_name == "opensearch_query" and "num_results" in entry:
282		print(f" Results: {entry['num_results']}")
283
284		print("=" * 60)
285
286		def get_directory_structure(self, path: str = '.', max_depth: int = 2) -> Dict[str, Any]:
287		"""
288		Get a formatted directory structure.
289
290		Args:
291		path: Starting directory path
292		max_depth: Maximum directory depth to show
293
294		Returns:
295		dict: Directory structure information
296		"""
297		try:
298		self.print_status(f"Getting directory structure for: {path} (depth: {max_depth})")
299
300		# Execute find command to get the structure
301		from ollamatools.tools.system_tools import execute_command
302		result = execute_command(
303		command="find",
304		args=[path, "-maxdepth", str(max_depth), "-type", "f", "-o", "-type", "d"],
305		permission_manager=self.agent.permission_manager
306		)
307
308		if "error" in result:
309		return result
310
311		# Process the output into a more structured format
312		lines = result.get("stdout", "").strip().split('\n')
313
314		structure = {
315		"path": path,
316		"depth": max_depth,
317		"entries": lines,
318		"files": [line for line in lines if os.path.isfile(line)],
319		"directories": [line for line in lines if os.path.isdir(line)],
320		"count": {
321		"total": len(lines),
322		"files": len([line for line in lines if os.path.isfile(line)]),
323		"directories": len([line for line in lines if os.path.isdir(line)])
324		}
325		}
326
327		self.print_success(f"Found {structure['count']['total']} entries "
328		f"({structure['count']['files']} files, "
329		f"{structure['count']['directories']} directories)")
330
331		return structure
332
333		except Exception as e:
334		self.print_error(f"Failed to get directory structure: {str(e)}")
335		self.suggest_fix(self._generate_error_fix_suggestion(e))
336
337		return {
338		"error": str(e),
339		"path": path,
340		"depth": max_depth
341		}
342
343		def _generate_error_fix_suggestion(self, error: Exception) -> str:
344		"""
345		Generate a fix suggestion based on the error type.
346
347		Args:
348		error: The exception
349
350		Returns:
351		str: Suggested fix
352		"""
353		if isinstance(error, CommandError):
354		return self._generate_command_fix_suggestion(error)
355		elif isinstance(error, FileError):
356		return self._generate_file_fix_suggestion(error)
357		elif isinstance(error, WebError):
358		return self._generate_web_fix_suggestion(error)
359		else:
360		# Generic suggestion
361		return f"Try a different approach or check if the required resources are available."
362
363		def _generate_command_fix_suggestion(self, error: Union[CommandError, Dict[str, Any]]) -> str:
364		"""
365		Generate a fix suggestion for command errors.
366
367		Args:
368		error: Error object or dictionary
369
370		Returns:
371		str: Suggested fix
372		"""
373		# Extract error details
374		if isinstance(error, CommandError):
375		error_msg = str(error)
376		details = error.details or {}
377		command = details.get('command', '')
378		stderr = details.get('stderr', '')
379		else:
380		error_msg = error.get('error', '')
381		command = error.get('command', '')
382		stderr = error.get('stderr', '')
383
384		# Check for common error patterns
385		if "not in allowed command list" in error_msg:
386		return (f"The command '{command}' is not in the allowed command list. "
387		f"Try using one of these commands instead: {', '.join(sorted(list(self.command_allowlist)[:10]))}... "
388		f"or add this command to the allowlist when initializing the agent.")
389
390		elif "dangerous pattern" in error_msg:
391		pattern = error_msg.split("pattern matching '")[1].split("'")[0] if "pattern matching '" in error_msg else ""
392		return (f"The command contains potentially dangerous pattern: {pattern}. "
393		f"Try reformulating the command to avoid shell metacharacters.")
394
395		elif "not found" in error_msg or "command not found" in stderr:
396		return f"The command '{command}' does not exist on this system. Check the spelling or install the command."
397
398		elif "permission denied" in stderr.lower():
399		return f"Permission denied when executing '{command}'. Try running with appropriate permissions."
400
401		else:
402		return f"Try a different command syntax or check command documentation."
403
404		def _generate_file_fix_suggestion(self, error: FileError) -> str:
405		"""
406		Generate a fix suggestion for file errors.
407
408		Args:
409		error: File error
410
411		Returns:
412		str: Suggested fix
413		"""
414		error_msg = str(error)
415		details = error.details or {}
416		path = details.get('path', '')
417
418		if "not found" in error_msg:
419		return f"The file '{path}' does not exist. Check the path or create the file."
420
421		elif "is a directory" in error_msg:
422		return f"'{path}' is a directory, not a file. Try specifying a file path instead."
423
424		elif "permission denied" in error_msg.lower():
425		return f"Permission denied when accessing '{path}'. Check file permissions."
426
427		elif "exceeds maximum size" in error_msg:
428		return f"The file '{path}' is too large. Try reading a smaller file or increasing max_size."
429
430		elif "outside the allowed directories" in error_msg:
431		return f"The path '{path}' is outside the allowed directories. Try a path within the sandbox."
432
433		else:
434		return f"Check if the file exists and you have appropriate permissions."
435
436		def _generate_web_fix_suggestion(self, error: WebError) -> str:
437		"""
438		Generate a fix suggestion for web errors.
439
440		Args:
441		error: Web error
442
443		Returns:
444		str: Suggested fix
445		"""
446		error_msg = str(error)
447		details = error.details or {}
448		url = details.get('url', '')
449
450		if "domain" in error_msg and "blocked" in error_msg:
451		return f"The domain in '{url}' is blocked for security reasons. Try a different domain."
452
453		elif "scheme" in error_msg and "not allowed" in error_msg:
454		return f"The URL scheme is not allowed. Use http:// or https:// instead."
455
456		elif "Connection refused" in error_msg:
457		return f"Could not connect to '{url}'. Check if the service is running and accessible."
458
459		elif "timeout" in error_msg.lower():
460		return f"The request to '{url}' timed out. Try again later or check your internet connection."
461
462		elif "Status code:" in error_msg and "404" in error_msg:
463		return f"The resource at '{url}' was not found (404). Check the URL for typos."
464
465		elif "Status code:" in error_msg and "403" in error_msg:
466		return f"Access to '{url}' is forbidden (403). The site may require authentication."
467
468		else:
469		return f"Try a different URL or check your internet connection."
470
471		def detect_command_intent(self, message: str) -> Tuple[bool, Optional[str]]:
472		"""
473		Detect if a message is asking for system information that requires a command.
474
475		Args:
476		message: User message
477
478		Returns:
479		Tuple[bool, Optional[str]]: (has_command_intent, suggested_command)
480		"""
481		return detect_command_intent(message, self.function_feedback)
482
483		def provide_function_feedback(self, query: str, correct_function: str, actual_function: str = None) -> bool:
484		"""
485		Allow users to provide feedback on function selection.
486
487		Args:
488		query: The original query
489		correct_function: The function that should have been used
490		actual_function: The function that was actually used (if known)
491
492		Returns:
493		bool: True if feedback was recorded successfully
494		"""
495		if actual_function:
496		# Record explicit feedback
497		self.function_feedback.record_explicit_feedback(
498		query, correct_function, actual_function, user_correction=True
499		)
500		self.print_success(f"Thank you for your feedback! I'll remember to use {correct_function} for similar queries in the future.")
501		return True
502		else:
503		# Try to find the most recent function used for this query
504		for entry in reversed(self.execution_history):
505		if entry.get("query", "") == query:
506		actual_function = entry.get("function")
507		if actual_function and actual_function != correct_function:
508		self.function_feedback.record_explicit_feedback(
509		query, correct_function, actual_function, user_correction=True
510		)
511		self.print_success(f"Thank you for your feedback! I'll remember to use {correct_function} for similar queries in the future.")
512		return True
513
514		# If we couldn't find a matching entry, just record the correct function
515		self.function_feedback.record_explicit_feedback(
516		query, correct_function, "unknown", user_correction=True
517		)
518		self.print_success(f"Thank you for your feedback! I'll remember to use {correct_function} for similar queries in the future.")
519		return True
520
521		def chat(self, message: str) -> str:
522		"""
523		Process a chat message, showing the execution steps transparently.
524
525		Args:
526		message: User message
527
528		Returns:
529		str: Response message
530		"""
531		try:
532		# First check if the message is asking for system information
533		has_command_intent, suggested_command = self.detect_command_intent(message)
534		if has_command_intent and suggested_command:
535		self.print_info(f"This appears to be a question about system information. Using '{suggested_command}' command...")
536		try:
537		# Execute the command
538		result = execute_command(
539		command=suggested_command,
540		permission_manager=self.agent.permission_manager
541		)
542
543		# Record this in execution history
544		self.execution_history.append({
545		"function": "execute_command",
546		"args": {"command": suggested_command},
547		"timestamp": time.time(),
548		"success": "error" not in result,
549		"returncode": result.get("returncode"),
550		"query": message
551		})
552
553		# Record implicit feedback
554		self.function_feedback.record_implicit_feedback(
555		message, "execute_command", 1.0, result
556		)
557
558		if "error" in result:
559		self.print_error(f"Command execution failed: {result['error']}")
560		return f"I tried to run the '{suggested_command}' command to answer your question, but encountered an error: {result['error']}"
561		else:
562		# Format the response based on the command
563		formatted_result = format_command_result(suggested_command, result)
564		return formatted_result
565		except Exception as e:
566		self.print_error(f"Command execution failed: {str(e)}")
567		return f"I tried to run the '{suggested_command}' command to answer your question, but encountered an error: {str(e)}"
568
569		# If not a command, check if the message likely requires a search but doesn't explicitly ask for one
570		if not message.lower().startswith('search '):
571		search_indicators = [
572		"what is", "how to", "who is", "where can", "when did", "why does",
573		"recipe for", "information about", "tell me about", "find", "look up",
574		"latest", "news on", "current", "recent"
575		]
576
577		# Check if the message contains search indicators
578		needs_search = any(indicator in message.lower() for indicator in search_indicators)
579
580		# Check for question structure
581		question_structure = message.lower().startswith(("what", "how", "who", "where", "when", "why", "can", "could"))
582
583		if needs_search or question_structure:
584		# Check if message is asking for external information we likely don't have
585		factual_intent = self.detect_factual_intent(message)
586		if factual_intent:
587		self.print_info("This appears to be a question that might benefit from search. Searching automatically...")
588		try:
589		results = self.execute_search(message)
590		# Process search results before continuing with normal chat flow
591		context = self.extract_search_context(results)
592		# Add context to the user message
593		message = f"{message}\n\nSearch context: {context}"
594
595		# Record implicit feedback
596		self.function_feedback.record_implicit_feedback(
597		message, "opensearch_query", 0.9, results
598		)
599		except Exception as e:
600		self.print_error(f"Automatic search failed: {str(e)}")
601		self.print_info("Continuing without search results")
602
603		# Print what we're doing
604		self.print_status(f"Processing message: {message}")
605
606		# Start timer for perf measurement
607		start_time = time.time()
608
609		# Process LLM response
610		try:
611		response = self.agent.chat(
612		message=message,
613		system_prompt=self.system_prompt,
614		context=self.context
615		)
616
617		# Check if the response looks like a JSON function call
618		if response.strip().startswith('{') and ('function' in response or 'name' in response):
619		self.print_info("Detected potential function call in response")
620		try:
621		# Parse the JSON response
622		func_data = json.loads(response)
623
624		# Extract function name and args
625		function_name = None
626		args = {}
627
628		# Handle different JSON formats
629		if 'function' in func_data:
630		function_name = func_data['function'].get('name')
631		args_str = func_data['function'].get('arguments', '{}')
632		if isinstance(args_str, str):
633		args = json.loads(args_str)
634		else:
635		args = args_str
636		elif 'name' in func_data:
637		function_name = func_data.get('name')
638		args = func_data.get('arguments', {})
639
640		if function_name:
641		self.print_status(f"Executing function: {function_name} with args: {args}")
642
643		# Execute the function
644		if function_name == 'opensearch_query' or function_name == 'search_web':
645		results = self.execute_search(args.get('query', message))
646		context = self.extract_search_context(results)
647		response = f"Here's what I found about '{args.get('query', message)}':\n\n{context}"
648		elif hasattr(self, function_name) and callable(getattr(self, function_name)):
649		# Execute a method of this class
650		method = getattr(self, function_name)
651		result = method(**args)
652		response = f"Here are the results for '{function_name}':\n\n{result}"
653		elif function_name in self.agent.registered_functions:
654		# Execute a registered function
655		result = self.agent.registered_functions[function_name](**args)
656		response = f"Here are the results for '{function_name}':\n\n{result}"
657		else:
658		self.print_error(f"Function '{function_name}' not found")
659		response = f"I tried to call a function '{function_name}' but it's not available. I'll try to help you directly instead."
660		except json.JSONDecodeError:
661		self.print_info("Response looked like JSON but couldn't be parsed as a function call")
662		except Exception as e:
663		self.print_error(f"Error executing function from response: {str(e)}")
664		response = f"I tried to execute a function to help with your request, but encountered an error: {str(e)}"
665		except Exception as e:
666		self.print_error(f"Error communicating with Ollama: {str(e)}")
667		response = f"I encountered an error communicating with the language model: {str(e)}"
668
669		# Update context
670		self.context.append({"role": "user", "content": message})
671		self.context.append({"role": "assistant", "content": response})
672
673		# Limit context size
674		if len(self.context) > 10:
675		self.context = self.context[-10:]
676
677		# Calculate execution time
678		execution_time = time.time() - start_time
679		self.print_success(f"Response generated in {execution_time:.2f} seconds")
680
681		return response
682
683		except OllamaToolsError as e:
684		error_details = handle_exception(e, log_error=True)
685
686		self.print_error(f"Error: {error_details['error_message']}")
687		self.suggest_fix(self._generate_error_fix_suggestion(e))
688
689		return f"I encountered an error: {error_details['error_message']}\n\n" + \
690		f"Suggested fix: {self._generate_error_fix_suggestion(e)}"
691
692		except Exception as e:
693		error_details = handle_exception(e, log_error=True)
694
695		self.print_error(f"Unexpected error: {str(e)}")
696
697		return f"I encountered an unexpected error: {str(e)}"

ollamatools-project/ollamatools/agent/transparent/commands.py +0 -122

…	…	@@ -1,122 +0,0 @@
1		"""
2		Command management functionality for the Transparent OllamaAgent.
3
4		This module provides functions for managing the command allowlist,
5		including listing, adding, and removing commands.
6		"""
7
8		from typing import List, Set
9		from .utils import print_info, print_error, print_success, print_warning
10
11		def list_allowed_commands(command_allowlist: Set[str]) -> List[str]:
12		"""
13		List all currently allowed commands.
14
15		Args:
16		command_allowlist: Set of currently allowed commands
17
18		Returns:
19		List[str]: List of allowed commands
20		"""
21		# Get the current allowlist
22		all_commands = sorted(list(command_allowlist))
23
24		# Print the commands
25		print_info(f"Currently allowed commands ({len(all_commands)}):")
26
27		# Format into categories for better readability
28		categories = {
29		"File Operations": ["ls", "cat", "find", "grep", "head", "tail"],
30		"Directory Operations": ["pwd", "cd", "mkdir", "rmdir"],
31		"System Information": ["uname", "hostname", "whoami", "ps", "top", "df", "du", "free"],
32		"Network Utilities": ["ping", "curl", "wget", "netstat", "dig", "nslookup"],
33		"Text Processing": ["echo", "printf", "sed", "awk", "cut", "sort", "uniq", "wc"],
34		"File Manipulation": ["cp", "mv", "rm", "touch", "chmod", "chown", "chgrp", "ln"],
35		"Other": []
36		}
37
38		# Categorize commands
39		for cmd in all_commands:
40		categorized = False
41		for category, commands in categories.items():
42		if cmd in commands:
43		categorized = True
44		break
45		if not categorized:
46		categories["Other"].append(cmd)
47
48		# Print by category
49		for category, commands in categories.items():
50		if commands:
51		print(f"\n{category}:")
52		for i, cmd in enumerate(sorted(commands)):
53		print(f" {cmd}", end="\t")
54		# New line every 5 commands for readability
55		if (i + 1) % 5 == 0:
56		print()
57		print() # Ensure we end with a newline
58
59		return all_commands
60
61		def add_allowed_command(command: str, command_allowlist: Set[str]) -> bool:
62		"""
63		Add a command to the allowed list.
64
65		Args:
66		command: Command to add
67		command_allowlist: Set of currently allowed commands
68
69		Returns:
70		bool: True if added successfully
71		"""
72		if not command or not isinstance(command, str):
73		print_error("Invalid command name")
74		return False
75
76		# Clean up the command name
77		command = command.strip().lower()
78
79		# Check if already in allowlist
80		if command in command_allowlist:
81		print_info(f"Command '{command}' is already allowed")
82		return True
83
84		# Add to custom allowlist
85		command_allowlist.add(command)
86		print_success(f"Command '{command}' added to allowed list")
87
88		# Show warning for potentially dangerous commands
89		dangerous_commands = {"rm", "mv", "chmod", "chown", "dd", "mkfs", "iptables"}
90		if command in dangerous_commands:
91		print_warning(f"'{command}' is a potentially dangerous command. Use with caution!")
92
93		return True
94
95		def remove_allowed_command(command: str, command_allowlist: Set[str]) -> bool:
96		"""
97		Remove a command from the allowed list.
98
99		Args:
100		command: Command to remove
101		command_allowlist: Set of currently allowed commands
102
103		Returns:
104		bool: True if removed successfully
105		"""
106		if not command or not isinstance(command, str):
107		print_error("Invalid command name")
108		return False
109
110		# Clean up the command name
111		command = command.strip().lower()
112
113		# Check if in allowlist
114		if command not in command_allowlist:
115		print_error(f"Command '{command}' is not in the allowed list")
116		return False
117
118		# Remove from custom allowlist
119		command_allowlist.remove(command)
120		print_success(f"Command '{command}' removed from allowed list")
121
122		return True

ollamatools-project/ollamatools/agent/transparent/feedback.py +0 -586

…	…	@@ -1,586 +0,0 @@
1		"""
2		Feedback system for the Transparent OllamaAgent.
3
4		This module provides a feedback loop system that improves the agent's
5		semantic understanding of user queries by learning from past interactions.
6		It helps map natural language queries to appropriate functions and commands.
7		"""
8
9		import os
10		import re
11		import json
12		import time
13		import logging
14		from typing import Dict, Any, List, Optional, Tuple, Set, Union
15
16		# Set up logging
17		logger = logging.getLogger(__name__)
18
19		class FunctionSelectionFeedback:
20		"""Manages feedback on function selection for continuous improvement."""
21
22		def __init__(self, feedback_file_path=None):
23		"""
24		Initialize the feedback system.
25
26		Args:
27		feedback_file_path: Path to the feedback file (default: ~/.ollamatools/function_selection_feedback.jsonl)
28		"""
29		if feedback_file_path is None:
30		self.feedback_file_path = os.path.join(
31		os.path.expanduser('~'),
32		'.ollamatools',
33		'function_selection_feedback.jsonl'
34		)
35		else:
36		self.feedback_file_path = feedback_file_path
37
38		# Create directory if it doesn't exist
39		os.makedirs(os.path.dirname(self.feedback_file_path), exist_ok=True)
40
41		# Cache of recent feedback for quick lookup
42		self.recent_feedback = []
43		self.max_cache_size = 1000
44
45		# Load existing feedback into cache
46		self._load_recent_feedback()
47
48		def _load_recent_feedback(self):
49		"""Load recent feedback entries into memory cache."""
50		try:
51		if os.path.exists(self.feedback_file_path):
52		with open(self.feedback_file_path, 'r') as f:
53		# Load the most recent entries (up to max_cache_size)
54		lines = f.readlines()
55		for line in lines[-self.max_cache_size:]:
56		try:
57		entry = json.loads(line.strip())
58		self.recent_feedback.append(entry)
59		except json.JSONDecodeError:
60		continue
61		logger.info(f"Loaded {len(self.recent_feedback)} feedback entries from {self.feedback_file_path}")
62		else:
63		logger.info(f"No feedback file found at {self.feedback_file_path}, starting with empty feedback")
64		except Exception as e:
65		logger.error(f"Error loading feedback data: {str(e)}")
66
67		def record_explicit_feedback(self, query, intended_function, actual_function, user_correction=True):
68		"""
69		Record explicit feedback when a function selection was incorrect.
70
71		Args:
72		query: The original user query
73		intended_function: The function that should have been used
74		actual_function: The function that was actually used
75		user_correction: Whether this feedback came directly from a user
76
77		Returns:
78		dict: The recorded feedback entry
79		"""
80		feedback_entry = {
81		"query": query,
82		"intended_function": intended_function,
83		"actual_function": actual_function,
84		"timestamp": time.time(),
85		"source": "user_correction" if user_correction else "system_detection",
86		"similarity_score": self._calculate_query_similarity(query)
87		}
88
89		# Add to cache
90		self.recent_feedback.append(feedback_entry)
91		if len(self.recent_feedback) > self.max_cache_size:
92		self.recent_feedback.pop(0)
93
94		# Write to file
95		try:
96		with open(self.feedback_file_path, 'a') as f:
97		f.write(json.dumps(feedback_entry) + '\n')
98		logger.info(f"Recorded explicit feedback for query: {query}")
99		except Exception as e:
100		logger.error(f"Error recording feedback: {str(e)}")
101
102		return feedback_entry
103
104		def record_implicit_feedback(self, query, function, confidence_score, execution_result):
105		"""
106		Record implicit feedback based on execution results.
107
108		Args:
109		query: The original user query
110		function: The function that was used
111		confidence_score: Confidence score for the function selection (0-1)
112		execution_result: Result of executing the function
113
114		Returns:
115		dict: The recorded feedback entry
116		"""
117		# Determine if the execution was likely successful
118		success_indicators = {
119		"execute_command": lambda r: r.get("returncode", 1) == 0 and not r.get("stderr"),
120		"read_file": lambda r: not (isinstance(r, str) and (r.startswith("Error") or r.startswith("Path validation error"))),
121		"search_web": lambda r: len(r) > 0 if isinstance(r, list) else True
122		}
123
124		likely_success = False
125		if function in success_indicators:
126		likely_success = success_indicators[function](execution_result)
127
128		feedback_entry = {
129		"query": query,
130		"function": function,
131		"timestamp": time.time(),
132		"source": "implicit_feedback",
133		"confidence_score": confidence_score,
134		"likely_success": likely_success,
135		"similarity_score": self._calculate_query_similarity(query)
136		}
137
138		# Only record if confidence was low or execution likely failed
139		if confidence_score < 0.7 or not likely_success:
140		# Add to cache
141		self.recent_feedback.append(feedback_entry)
142		if len(self.recent_feedback) > self.max_cache_size:
143		self.recent_feedback.pop(0)
144
145		# Write to file
146		try:
147		with open(self.feedback_file_path, 'a') as f:
148		f.write(json.dumps(feedback_entry) + '\n')
149		logger.info(f"Recorded implicit feedback for query: {query}")
150		except Exception as e:
151		logger.error(f"Error recording feedback: {str(e)}")
152
153		return feedback_entry
154
155		def _calculate_query_similarity(self, query):
156		"""
157		Calculate similarity between this query and previous queries.
158
159		Args:
160		query: The query to compare
161
162		Returns:
163		list: List of similar queries with their similarity scores
164		"""
165		# Simple word-based similarity for now
166		query_words = set(query.lower().split())
167
168		similarities = []
169		for entry in self.recent_feedback:
170		past_query = entry.get("query", "")
171		past_words = set(past_query.lower().split())
172
173		# Calculate Jaccard similarity
174		if not query_words or not past_words:
175		similarity = 0
176		else:
177		intersection = len(query_words.intersection(past_words))
178		union = len(query_words.union(past_words))
179		similarity = intersection / union if union > 0 else 0
180
181		if similarity > 0.3: # Only include reasonably similar queries
182		similarities.append({
183		"query": past_query,
184		"similarity": similarity,
185		"intended_function": entry.get("intended_function") or entry.get("function"),
186		"timestamp": entry.get("timestamp")
187		})
188
189		# Sort by similarity score
190		similarities.sort(key=lambda x: x["similarity"], reverse=True)
191		return similarities[:5] # Return top 5 similar queries
192
193		def get_function_suggestion(self, query):
194		"""
195		Get function suggestion based on feedback history.
196
197		Args:
198		query: The user query
199
200		Returns:
201		tuple: (suggested_function, confidence_score, similar_queries)
202		"""
203		# Get similar queries
204		similar_queries = self._calculate_query_similarity(query)
205
206		if not similar_queries:
207		return None, 0, []
208
209		# Count function occurrences weighted by similarity
210		function_scores = {}
211		for item in similar_queries:
212		function = item["intended_function"]
213		similarity = item["similarity"]
214
215		if function not in function_scores:
216		function_scores[function] = 0
217
218		function_scores[function] += similarity
219
220		# Find the function with the highest score
221		best_function = None
222		highest_score = 0
223
224		for function, score in function_scores.items():
225		if score > highest_score:
226		highest_score = score
227		best_function = function
228
229		# Calculate confidence based on the highest score and number of similar queries
230		confidence = highest_score / len(similar_queries) if similar_queries else 0
231
232		return best_function, confidence, similar_queries
233
234		def generate_improved_patterns(self):
235		"""
236		Generate improved regex patterns based on feedback data.
237
238		Returns:
239		dict: Dictionary mapping regex patterns to functions
240		"""
241		# Analyze the feedback data
242		analysis = self.analyze_function_selection_patterns()
243
244		if "error" in analysis:
245		return {}
246
247		improved_patterns = {}
248
249		# Process mismatches to generate patterns
250		for mismatch_info in analysis.get("top_mismatches", []):
251		mismatch, data = mismatch_info
252		actual, intended = mismatch.split(" → ")
253
254		# Skip if we don't have enough examples
255		if len(data["examples"]) < 2:
256		continue
257
258		# Generate a pattern from the examples
259		pattern = self._generate_regex_from_examples(data["examples"], intended)
260		if pattern:
261		improved_patterns[pattern] = intended
262
263		# Process query patterns
264		for pattern_info in analysis.get("top_query_patterns", []):
265		pattern_text, data = pattern_info
266		functions = data["functions"]
267
268		if not functions:
269		continue
270
271		# Find the most common function
272		most_common = max(functions.items(), key=lambda x: x[1])
273		function_name, count = most_common
274
275		# Only include if it's a strong pattern (used for the same function >70% of the time)
276		confidence = count / data["count"]
277		if confidence > 0.7:
278		# Create a regex pattern from the text pattern
279		regex_pattern = r"\b" + re.escape(pattern_text) + r".*"
280		improved_patterns[regex_pattern] = function_name
281
282		return improved_patterns
283
284		def _generate_regex_from_examples(self, examples, function):
285		"""
286		Generate a regex pattern from example queries.
287
288		Args:
289		examples: List of example queries
290		function: The target function
291
292		Returns:
293		str: Regex pattern
294		"""
295		if not examples:
296		return None
297
298		# Extract common words across examples
299		common_words = set(examples[0].lower().split())
300		for example in examples[1:]:
301		common_words &= set(example.lower().split())
302
303		# Remove very common words
304		stop_words = {"what", "is", "the", "a", "an", "in", "on", "at", "to", "for", "of", "with", "by"}
305		common_words -= stop_words
306
307		if not common_words:
308		# If no common words, try to find common starting words
309		starts_with = []
310		for example in examples:
311		words = example.lower().split()
312		if words:
313		starts_with.append(words[0])
314
315		# If we have consistent starting words
316		if len(set(starts_with)) == 1 and starts_with[0] not in stop_words:
317		return r"\b" + re.escape(starts_with[0]) + r".*"
318
319		return None
320
321		# Create a pattern with the common words
322		pattern_parts = []
323		for word in common_words:
324		pattern_parts.append(r"\b" + re.escape(word) + r"\b")
325
326		# Combine with lookaheads for flexibility in word order
327		pattern = "".join(f"(?=.{part})" for part in pattern_parts) + "."
328		return pattern
329
330		def analyze_function_selection_patterns(self):
331		"""
332		Analyze patterns in function selection feedback.
333
334		Returns:
335		dict: Analysis results
336		"""
337		try:
338		# Load all feedback data
339		all_feedback = []
340		if os.path.exists(self.feedback_file_path):
341		with open(self.feedback_file_path, 'r') as f:
342		for line in f:
343		try:
344		entry = json.loads(line.strip())
345		all_feedback.append(entry)
346		except json.JSONDecodeError:
347		continue
348
349		if not all_feedback:
350		return {"error": "No feedback data available"}
351
352		# Analyze common mismatches
353		mismatches = {}
354		for entry in all_feedback:
355		if "intended_function" in entry and "actual_function" in entry:
356		if entry["intended_function"] != entry["actual_function"]:
357		key = f"{entry['actual_function']} → {entry['intended_function']}"
358		if key not in mismatches:
359		mismatches[key] = {
360		"count": 0,
361		"examples": []
362		}
363
364		mismatches[key]["count"] += 1
365		if len(mismatches[key]["examples"]) < 5: # Limit examples
366		mismatches[key]["examples"].append(entry["query"])
367
368		# Sort mismatches by frequency
369		sorted_mismatches = sorted(
370		[(k, v) for k, v in mismatches.items()],
371		key=lambda x: x[1]["count"],
372		reverse=True
373		)
374
375		# Analyze query patterns
376		query_patterns = {}
377		for entry in all_feedback:
378		query = entry.get("query", "")
379		words = query.lower().split()
380
381		# Extract potential patterns (first 2-3 words)
382		if len(words) >= 2:
383		pattern = " ".join(words[:2])
384		if pattern not in query_patterns:
385		query_patterns[pattern] = {
386		"count": 0,
387		"functions": {}
388		}
389
390		query_patterns[pattern]["count"] += 1
391
392		function = entry.get("intended_function") or entry.get("function")
393		if function:
394		if function not in query_patterns[pattern]["functions"]:
395		query_patterns[pattern]["functions"][function] = 0
396		query_patterns[pattern]["functions"][function] += 1
397
398		# Sort patterns by frequency
399		sorted_patterns = sorted(
400		[(k, v) for k, v in query_patterns.items()],
401		key=lambda x: x[1]["count"],
402		reverse=True
403		)
404
405		return {
406		"total_feedback_entries": len(all_feedback),
407		"top_mismatches": sorted_mismatches[:10],
408		"top_query_patterns": sorted_patterns[:10],
409		"improvement_suggestions": self._generate_improvement_suggestions(sorted_mismatches, sorted_patterns)
410		}
411		except Exception as e:
412		logger.error(f"Error analyzing feedback: {str(e)}")
413		return {"error": f"Error analyzing feedback: {str(e)}"}
414
415		def _generate_improvement_suggestions(self, mismatches, patterns):
416		"""
417		Generate suggestions for improving function selection.
418
419		Args:
420		mismatches: List of function mismatches
421		patterns: List of query patterns
422
423		Returns:
424		list: Improvement suggestions
425		"""
426		suggestions = []
427
428		# Suggest improvements based on common mismatches
429		if mismatches:
430		for mismatch, data in mismatches[:5]:
431		actual, intended = mismatch.split(" → ")
432		examples = data["examples"]
433
434		if examples:
435		pattern_words = set()
436		for example in examples:
437		pattern_words.update(example.lower().split())
438
439		# Filter out common words
440		common_words = {"what", "is", "the", "a", "an", "in", "on", "at", "to", "for", "of", "with", "by"}
441		pattern_words = pattern_words - common_words
442
443		if pattern_words:
444		suggestion = {
445		"type": "function_mapping",
446		"description": f"Improve mapping from '{actual}' to '{intended}'",
447		"examples": examples[:3],
448		"suggested_keywords": list(pattern_words)[:5],
449		"implementation_hint": f"Add pattern matching for these keywords to select '{intended}'"
450		}
451		suggestions.append(suggestion)
452
453		# Suggest improvements based on query patterns
454		if patterns:
455		for pattern, data in patterns[:5]:
456		functions = data["functions"]
457		if functions:
458		# Find the most common function for this pattern
459		most_common = max(functions.items(), key=lambda x: x[1])
460		function_name, count = most_common
461
462		suggestion = {
463		"type": "query_pattern",
464		"description": f"Add explicit handling for queries starting with '{pattern}'",
465		"suggested_function": function_name,
466		"confidence": count / data["count"],
467		"implementation_hint": f"Add pattern matching for queries starting with '{pattern}' to select '{function_name}'"
468		}
469		suggestions.append(suggestion)
470
471		return suggestions
472
473		def detect_command_intent(message, feedback_system=None):
474		"""
475		Detect if a message is asking for system information that requires a command.
476
477		Args:
478		message: User message
479		feedback_system: Optional FunctionSelectionFeedback instance
480
481		Returns:
482		Tuple[bool, Optional[str]]: (has_command_intent, suggested_command)
483		"""
484		# First check feedback-based patterns if available
485		if feedback_system:
486		suggested_function, confidence, similar_queries = feedback_system.get_function_suggestion(message)
487
488		if suggested_function == "execute_command" and confidence > 0.7:
489		# Extract command from similar queries
490		command = extract_command_from_similar_queries(message, similar_queries)
491		if command:
492		return True, command
493
494		# Map common questions to shell commands
495		command_patterns = {
496		r"(?:what\|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+directory": "pwd",
497		r"where\s+am\s+i": "pwd",
498		r"list\s+(?:the\s+)?(?:files\|contents)": "ls",
499		r"show\s+(?:the\s+)?(?:files\|contents)": "ls",
500		r"(?:what\|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+user": "whoami",
501		r"who\s+am\s+i": "whoami",
502		r"(?:what\|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+date": "date",
503		r"(?:what\|which)(?:\s+is)?(?:\s+the)?(?:\s+current)?\s+time": "date",
504		# Add more patterns as needed
505		}
506
507		# Check if message matches any pattern
508		for pattern, command in command_patterns.items():
509		if re.search(pattern, message.lower()):
510		return True, command
511
512		return False, None
513
514		def extract_command_from_similar_queries(message, similar_queries):
515		"""
516		Extract the most appropriate command from similar queries.
517
518		Args:
519		message: Current user message
520		similar_queries: List of similar queries with their metadata
521
522		Returns:
523		str: Extracted command or None
524		"""
525		# Count command occurrences in similar queries
526		command_counts = {}
527
528		for query_info in similar_queries:
529		# Skip if not execute_command
530		if query_info.get("intended_function") != "execute_command":
531		continue
532
533		# Try to extract the command from the query
534		query = query_info.get("query", "")
535
536		# Check for common command patterns
537		command_mapping = {
538		r"directory\|where\s+am\s+i": "pwd",
539		r"list\s+files\|show\s+files\|contents": "ls",
540		r"user\|who\s+am\s+i": "whoami",
541		r"date\|time": "date"
542		}
543
544		for pattern, cmd in command_mapping.items():
545		if re.search(pattern, query.lower()):
546		if cmd not in command_counts:
547		command_counts[cmd] = 0
548		command_counts[cmd] += query_info.get("similarity", 0.5)
549
550		# Return the most common command
551		if command_counts:
552		return max(command_counts.items(), key=lambda x: x[1])[0]
553
554		return None
555
556		def format_command_result(command, result):
557		"""
558		Format the result of a command execution in a user-friendly way.
559
560		Args:
561		command: The command that was executed
562		result: The command execution result
563
564		Returns:
565		str: Formatted result
566		"""
567		if "error" in result:
568		return f"Error executing command: {result['error']}"
569
570		stdout = result.get("stdout", "").strip()
571		stderr = result.get("stderr", "").strip()
572
573		if command == "pwd":
574		return f"You are currently in: {stdout}"
575		elif command == "ls":
576		return f"Files in the current directory:\n{stdout}"
577		elif command == "whoami":
578		return f"Current user: {stdout}"
579		elif command == "date":
580		return f"Current date and time: {stdout}"
581		else:
582		# Generic formatting
583		response = f"Command result:\n{stdout}"
584		if stderr:
585		response += f"\nErrors/Warnings:\n{stderr}"
586		return response

ollamatools-project/ollamatools/agent/transparent/issues.py +0 -189

…	…	@@ -1,189 +0,0 @@
1		"""
2		Issue tracking functionality for the Transparent OllamaAgent.
3
4		This module provides functions for managing issues, including
5		loading, saving, adding, listing, and updating issues.
6		"""
7
8		import os
9		import json
10		import time
11		from typing import Dict, Any, List, Optional
12
13		from .utils import print_info, print_error, print_success
14
15		def load_issues(agent) -> None:
16		"""
17		Load saved issues from file.
18
19		Args:
20		agent: The agent instance
21		"""
22		if os.path.exists(agent.issues_file):
23		try:
24		with open(agent.issues_file, 'r') as f:
25		data = json.load(f)
26		agent.issues = data.get('issues', [])
27		agent.issue_id_counter = data.get('next_id', 1)
28		agent.print_info(f"Loaded {len(agent.issues)} issues from {agent.issues_file}")
29		except Exception as e:
30		agent.print_error(f"Failed to load issues: {str(e)}")
31		agent.issues = []
32		agent.issue_id_counter = 1
33
34		def save_issues(agent) -> None:
35		"""
36		Save issues to file.
37
38		Args:
39		agent: The agent instance
40		"""
41		try:
42		with open(agent.issues_file, 'w') as f:
43		json.dump({
44		'issues': agent.issues,
45		'next_id': agent.issue_id_counter
46		}, f, indent=2)
47		agent.print_info(f"Saved {len(agent.issues)} issues to {agent.issues_file}")
48		except Exception as e:
49		agent.print_error(f"Failed to save issues: {str(e)}")
50
51		def add_issue(agent, title: str, description: str, severity: str = 'medium') -> int:
52		"""
53		Add a new issue to the tracker.
54
55		Args:
56		agent: The agent instance
57		title: Issue title
58		description: Issue description
59		severity: Issue severity (low, medium, high)
60
61		Returns:
62		int: Issue ID
63		"""
64		# Validate severity
65		if severity not in ['low', 'medium', 'high']:
66		severity = 'medium'
67
68		# Create issue
69		issue = {
70		'id': agent.issue_id_counter,
71		'title': title,
72		'description': description,
73		'severity': severity,
74		'status': 'open',
75		'created_at': time.strftime('%Y-%m-%d %H:%M:%S'),
76		'updated_at': time.strftime('%Y-%m-%d %H:%M:%S')
77		}
78
79		# Add to list and increment counter
80		agent.issues.append(issue)
81		agent.issue_id_counter += 1
82
83		# Save issues
84		save_issues(agent)
85
86		agent.print_success(f"Issue #{issue['id']} added: {title}")
87		return issue['id']
88
89		def list_issues(agent, status: Optional[str] = None) -> List[Dict[str, Any]]:
90		"""
91		List all issues, optionally filtered by status.
92
93		Args:
94		agent: The agent instance
95		status: Optional status filter (open, resolved)
96
97		Returns:
98		List[Dict[str, Any]]: List of issues
99		"""
100		# Filter issues if status provided
101		filtered_issues = agent.issues
102		if status:
103		filtered_issues = [i for i in agent.issues if i['status'] == status]
104
105		# Print issues
106		if not filtered_issues:
107		agent.print_info("No issues found")
108		return []
109
110		agent.print_info(f"Issues ({len(filtered_issues)}):")
111		print("\nID \| Severity \| Status \| Title")
112		print("----\|----------\|----------\|-------------------")
113
114		for issue in filtered_issues:
115		print(f"{issue['id']:<4}\| {issue['severity']:<9}\| {issue['status']:<9}\| {issue['title']}")
116
117		return filtered_issues
118
119		def get_issue(agent, issue_id: int) -> Optional[Dict[str, Any]]:
120		"""
121		Get details of a specific issue.
122
123		Args:
124		agent: The agent instance
125		issue_id: Issue ID
126
127		Returns:
128		Optional[Dict[str, Any]]: Issue details or None if not found
129		"""
130		# Find issue
131		for issue in agent.issues:
132		if issue['id'] == issue_id:
133		# Print issue details
134		agent.print_info(f"Issue #{issue['id']}: {issue['title']}")
135		print(f"Status: {issue['status']}")
136		print(f"Severity: {issue['severity']}")
137		print(f"Created: {issue['created_at']}")
138		print(f"Updated: {issue['updated_at']}")
139		print("\nDescription:")
140		print(issue['description'])
141		return issue
142
143		agent.print_error(f"Issue #{issue_id} not found")
144		return None
145
146		def update_issue(
147		agent,
148		issue_id: int,
149		status: Optional[str] = None,
150		severity: Optional[str] = None,
151		description: Optional[str] = None
152		) -> bool:
153		"""
154		Update an existing issue.
155
156		Args:
157		agent: The agent instance
158		issue_id: Issue ID
159		status: New status (open, resolved)
160		severity: New severity (low, medium, high)
161		description: New description
162
163		Returns:
164		bool: True if updated successfully
165		"""
166		# Find issue
167		for i, issue in enumerate(agent.issues):
168		if issue['id'] == issue_id:
169		# Update fields
170		if status and status in ['open', 'resolved']:
171		agent.issues[i]['status'] = status
172
173		if severity and severity in ['low', 'medium', 'high']:
174		agent.issues[i]['severity'] = severity
175
176		if description:
177		agent.issues[i]['description'] = description
178
179		# Update timestamp
180		agent.issues[i]['updated_at'] = time.strftime('%Y-%m-%d %H:%M:%S')
181
182		# Save issues
183		save_issues(agent)
184
185		agent.print_success(f"Issue #{issue_id} updated")
186		return True
187
188		agent.print_error(f"Issue #{issue_id} not found")
189		return False

ollamatools-project/ollamatools/agent/transparent/search.py +0 -208

…	…	@@ -1,208 +0,0 @@
1		"""
2		Search functionality for the Transparent OllamaAgent.
3
4		This module provides search-related functionality for the TransparentOllamaAgent,
5		including natural language detection of search intent, search execution, and
6		search results processing.
7		"""
8
9		import time
10		import logging
11		from typing import Dict, Any, List, Optional, Set
12
13		from ollamatools.tools.web_tools import search_web
14		from ollamatools.errors import WebError
15		from ollamatools.opensearch_web_tools import opensearch_query
16		from .utils import print_status, print_success, print_error, print_warning, print_info
17
18		# Set up logger
19		logger = logging.getLogger(__name__)
20
21		def detect_factual_intent(message: str) -> bool:
22		"""
23		Detect if a message is likely asking for factual information
24		that would benefit from search.
25
26		Args:
27		message: User message
28
29		Returns:
30		bool: True if message likely needs external information
31		"""
32		# Basic topics that often require external information
33		external_topics = [
34		"recipe", "weather", "news", "sports", "movie", "book", "author",
35		"product", "review", "price", "stock", "company", "history",
36		"person", "celebrity", "politician", "event", "scientific",
37		"research", "study", "statistic", "data", "definition"
38		]
39
40		# Check if message contains any of these topics
41		contains_topic = any(topic in message.lower() for topic in external_topics)
42
43		# Check if it's a complex question (longer questions tend to be more specific)
44		is_complex = len(message.split()) > 6
45
46		# Simple heuristic: if it contains a topic of interest and is reasonably complex
47		return contains_topic and is_complex
48
49		def extract_search_context(search_results: Dict[str, Any]) -> str:
50		"""
51		Extract useful context from search results.
52
53		Args:
54		search_results: Results from opensearch_query
55
56		Returns:
57		str: Formatted context for the agent
58		"""
59		context_parts = []
60
61		# Add direct answer if available
62		if search_results.get("answer"):
63		try:
64		# Handle both string and dict answers
65		answer = search_results["answer"]
66		if isinstance(answer, dict):
67		answer_text = answer.get("content", str(answer))
68		else:
69		answer_text = str(answer)
70		context_parts.append(f"Direct answer: {answer_text}")
71		except Exception as e:
72		logger.warning(f"Error extracting answer for context: {str(e)}")
73
74		# Add snippets from results
75		if search_results.get("results"):
76		context_parts.append("Information from search results:")
77		for i, result in enumerate(search_results["results"][:3]): # Limit to 3 results
78		title = result.get("title", "Untitled")
79		content = result.get("content", "No content")
80		url = result.get("url", "")
81		context_parts.append(f"- {title}: {content} (Source: {url})")
82
83		# Join everything with newlines
84		return "\n".join(context_parts)
85
86		def execute_search(
87		query: str,
88		opensearch_config: Dict[str, Any],
89		permission_manager: Any,
90		execution_history: List[Dict[str, Any]]
91		) -> Dict[str, Any]:
92		"""
93		Execute a search using OpenSearchNG or fall back to standard search.
94
95		Args:
96		query: Search query
97		opensearch_config: OpenSearchNG configuration
98		permission_manager: Permission manager instance
99		execution_history: History of executed functions
100
101		Returns:
102		dict: Search results
103		"""
104		try:
105		print_status(f"Searching for: {query}")
106
107		# Try to use OpenSearchNG first
108		try:
109		# Execute the search with OpenSearchNG
110		results = opensearch_query(
111		query=query,
112		config=opensearch_config,
113		permission_manager=permission_manager
114		)
115
116		# Show the number of results
117		num_results = len(results.get("results", []))
118		print_success(f"Found {num_results} results using OpenSearchNG")
119
120		# Show suggested answer if available
121		if results.get("answer"):
122		try:
123		# Handle both string and dict answers
124		answer = results["answer"]
125		if isinstance(answer, dict):
126		answer_text = answer.get("content", str(answer))
127		else:
128		answer_text = str(answer)
129		print_info(f"Direct answer: {answer_text}")
130		except Exception as e:
131		logger.warning(f"Error displaying answer: {str(e)}")
132
133		# Register in execution history
134		execution_history.append({
135		"function": "opensearch_query",
136		"args": {"query": query},
137		"timestamp": time.time(),
138		"success": True,
139		"num_results": num_results
140		})
141
142		return results
143
144		except (WebError, ImportError, TypeError) as e:
145		# If OpenSearchNG fails, fall back to standard web search
146		if "Could not connect to OpenSearchNG server" in str(e) or isinstance(e, ImportError):
147		print_warning("OpenSearchNG is not available. Using fallback search method.")
148
149		# Use the fallback search method from web_tools
150		search_results = search_web(
151		query,
152		num_results=5,
153		permission_manager=permission_manager
154		)
155
156		# Convert results to the expected format
157		fallback_results = {
158		"query": query,
159		"results": [],
160		"suggestion": None,
161		"answer": None,
162		"engines": ["fallback"],
163		"timing": {"total": 0}
164		}
165
166		for result in search_results:
167		fallback_results["results"].append({
168		"title": result.get("title", ""),
169		"url": result.get("url", ""),
170		"content": result.get("snippet", ""),
171		"engine": "fallback"
172		})
173
174		num_results = len(fallback_results["results"])
175		print_success(f"Found {num_results} results using fallback search")
176
177		# Register in execution history
178		execution_history.append({
179		"function": "search_web_fallback",
180		"args": {"query": query},
181		"timestamp": time.time(),
182		"success": True,
183		"num_results": num_results
184		})
185
186		return fallback_results
187		else:
188		# Re-raise other WebError exceptions
189		raise
190
191		except Exception as e:
192		print_error(f"Search failed: {str(e)}")
193
194		# Register in execution history
195		execution_history.append({
196		"function": "execute_search",
197		"args": {"query": query},
198		"timestamp": time.time(),
199		"success": False,
200		"error": str(e)
201		})
202
203		# Return empty results
204		return {
205		"query": query,
206		"results": [],
207		"error": str(e)
208		}

ollamatools-project/ollamatools/agent/transparent/utils.py +0 -136

…	…	@@ -1,136 +0,0 @@
1		"""
2		Utility functions for the Transparent OllamaAgent.
3
4		This module provides utility functions for displaying status messages,
5		formatting output, and suggesting fixes for errors.
6		"""
7
8		def print_status(message: str) -> None:
9		"""
10		Print a status message.
11
12		Args:
13		message: Status message to print
14		"""
15		print(f"\n🔄 {message}")
16
17		def print_success(message: str) -> None:
18		"""
19		Print a success message.
20
21		Args:
22		message: Success message to print
23		"""
24		print(f"✅ {message}")
25
26		def print_error(message: str) -> None:
27		"""
28		Print an error message.
29
30		Args:
31		message: Error message to print
32		"""
33		print(f"❌ {message}")
34
35		def print_warning(message: str) -> None:
36		"""
37		Print a warning message.
38
39		Args:
40		message: Warning message to print
41		"""
42		print(f"⚠️ {message}")
43
44		def print_info(message: str) -> None:
45		"""
46		Print an information message.
47
48		Args:
49		message: Information message to print
50		"""
51		print(f"ℹ️ {message}")
52
53		def suggest_fix(suggestion: str) -> None:
54		"""
55		Print a fix suggestion.
56
57		Args:
58		suggestion: Fix suggestion to print
59		"""
60		print(f"💡 Suggested fix: {suggestion}")
61
62		def format_code_block(code: str, language: str = "python") -> str:
63		"""
64		Format text as a code block.
65
66		Args:
67		code: Code to format
68		language: Programming language for syntax highlighting
69
70		Returns:
71		str: Formatted code block
72		"""
73		return f"```{language}\n{code}\n```"
74
75		def format_table(headers: list, rows: list) -> str:
76		"""
77		Format data as a table.
78
79		Args:
80		headers: Table headers
81		rows: Table rows
82
83		Returns:
84		str: Formatted table
85		"""
86		# Determine column widths
87		col_widths = [len(h) for h in headers]
88		for row in rows:
89		for i, cell in enumerate(row):
90		col_widths[i] = max(col_widths[i], len(str(cell)))
91
92		# Create the header
93		header = "\| " + " \| ".join(h.ljust(w) for h, w in zip(headers, col_widths)) + " \|"
94		separator = "\|-" + "-\|-".join("-" * w for w in col_widths) + "-\|"
95
96		# Create the rows
97		formatted_rows = []
98		for row in rows:
99		formatted_row = "\| " + " \| ".join(str(cell).ljust(w) for cell, w in zip(row, col_widths)) + " \|"
100		formatted_rows.append(formatted_row)
101
102		# Combine everything
103		return "\n".join([header, separator] + formatted_rows)
104
105		def truncate_text(text: str, max_length: int = 100, suffix: str = "...") -> str:
106		"""
107		Truncate text to a maximum length.
108
109		Args:
110		text: Text to truncate
111		max_length: Maximum length
112		suffix: Suffix to add to truncated text
113
114		Returns:
115		str: Truncated text
116		"""
117		if len(text) <= max_length:
118		return text
119		return text[:max_length - len(suffix)] + suffix
120
121		def format_execution_time(seconds: float) -> str:
122		"""
123		Format execution time in a human-readable format.
124
125		Args:
126		seconds: Execution time in seconds
127
128		Returns:
129		str: Formatted execution time
130		"""
131		if seconds < 0.001:
132		return f"{seconds * 1000000:.2f}μs"
133		elif seconds < 1:
134		return f"{seconds * 1000:.2f}ms"
135		else:
136		return f"{seconds:.2f}s"

ollamatools-project/tests/test_natural_language_search.py +8 -22

…	…	@@ -2,7 +2,7 @@
2	2	"""
3	3	Test script for natural language search understanding and function execution.
4	4
5		This script tests the TransparentOllamaAgent's ability to:
	5	This script tests the ModularOllamaAgent's ability to:
6	6	1. Detect when a natural language query requires search
7	7	2. Properly execute function calls returned by the LLM
8	8	3. Handle search results correctly
…	…	@@ -20,23 +20,9 @@ from unittest.mock import patch, MagicMock
20	20	# Add the parent directory to the Python path to import ollamatools
21	21	sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
22	22
23		# Import directly from the module files using importlib
24		import importlib.util
25
26		# Load the agent module
27		agent_path = os.path.join(os.path.dirname(__file__), '..', 'ollamatools', 'agent', 'transparent', 'agent.py')
28		spec = importlib.util.spec_from_file_location("agent_module", agent_path)
29		agent_module = importlib.util.module_from_spec(spec)
30		spec.loader.exec_module(agent_module)
31		TransparentOllamaAgent = agent_module.TransparentOllamaAgent
32
33		# Load the search module
34		search_path = os.path.join(os.path.dirname(__file__), '..', 'ollamatools', 'agent', 'transparent', 'search.py')
35		spec = importlib.util.spec_from_file_location("search_module", search_path)
36		search_module = importlib.util.module_from_spec(spec)
37		spec.loader.exec_module(search_module)
38		detect_factual_intent = search_module.detect_factual_intent
39		extract_search_context = search_module.extract_search_context
	23	# Import the modular agent
	24	from ollamatools.agent import ModularOllamaAgent
	25	from ollamatools.agent.modular.search import detect_factual_intent, extract_search_context, execute_search
40	26
41	27	class TestNaturalLanguageSearch(unittest.TestCase):
42	28	"""Test cases for natural language search understanding."""
…	…	@@ -44,7 +30,7 @@ class TestNaturalLanguageSearch(unittest.TestCase):
44	30	def setUp(self):
45	31	"""Set up test environment."""
46	32	# Create a mock agent with mocked dependencies
47		self.agent = TransparentOllamaAgent(
	33	self.agent = ModularOllamaAgent(
48	34	model="llama3.1:latest",
49	35	permission_prompts=False
50	36	)
…	…	@@ -119,7 +105,7 @@ class TestNaturalLanguageSearch(unittest.TestCase):
119	105	self.assertIn("Homemade Chili Sauce Recipe", context)
120	106	self.assertIn("Spicy Chili Sauce - Quick and Easy", context)
121	107
122		@patch('ollamatools.agent.transparent.search.execute_search')
	108	@patch('ollamatools.agent.modular.search.execute_search')
123	109	def test_natural_language_search_trigger(self, mock_execute_search):
124	110	"""Test that natural language queries trigger search appropriately."""
125	111	# Mock the execute_search function to return sample results
…	…	@@ -140,7 +126,7 @@ class TestNaturalLanguageSearch(unittest.TestCase):
140	126	"What's a good recipe for chilli sauce?"
141	127	)
142	128
143		@patch('ollamatools.agent.transparent.agent.json.loads')
	129	@patch('ollamatools.agent.modular.agent.json.loads')
144	130	def test_function_call_detection(self, mock_json_loads):
145	131	"""Test detection and execution of function calls in LLM responses."""
146	132	# Mock the agent.chat method to return a function call
…	…	@@ -170,7 +156,7 @@ class TestNaturalLanguageSearch(unittest.TestCase):
170	156	self.assertNotIn('"function":', response)
171	157	self.assertIn("Here's what I found about", response)
172	158
173		@patch('ollamatools.agent.transparent.agent.json.loads')
	159	@patch('ollamatools.agent.modular.agent.json.loads')
174	160	def test_malformed_function_call_handling(self, mock_json_loads):
175	161	"""Test handling of malformed function calls."""
176	162	# Mock json.loads to raise JSONDecodeError

ollama_semantic_function_calling commits / ede5577

Changed files (16) +284 -3437